Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
724dc20202246d41362380aa4308035c83f9b793
[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 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(simix_network, simix, "SIMIX network-related synchronization");
15
16 static void SIMIX_mbox_free(void *data);
17 static xbt_dict_t mailboxes = xbt_dict_new_homogeneous(SIMIX_mbox_free);
18
19 static void SIMIX_waitany_remove_simcall_from_actions(smx_simcall_t simcall);
20 static void SIMIX_comm_copy_data(smx_synchro_t comm);
21 static smx_synchro_t SIMIX_comm_new(e_smx_comm_type_t type);
22 static inline void SIMIX_mbox_push(smx_mailbox_t mbox, smx_synchro_t comm);
23 static smx_synchro_t _find_matching_comm(std::deque<smx_synchro_t> *deque, e_smx_comm_type_t type,
24     int (*match_fun)(void *, void *,smx_synchro_t), void *user_data, smx_synchro_t my_synchro, bool remove_matching);
25 static void SIMIX_comm_start(smx_synchro_t synchro);
26
27 void SIMIX_mailbox_exit(void)
28 {
29   xbt_dict_free(&mailboxes);
30 }
31
32 /******************************************************************************/
33 /*                           Rendez-Vous Points                               */
34 /******************************************************************************/
35
36 smx_mailbox_t SIMIX_mbox_create(const char *name)
37 {
38   xbt_assert(name, "Mailboxes must have a name");
39   /* two processes may have pushed the same mbox_create simcall at the same time */
40   smx_mailbox_t mbox = (smx_mailbox_t) xbt_dict_get_or_null(mailboxes, name);
41
42   if (!mbox) {
43     mbox = xbt_new0(s_smx_mailbox_t, 1);
44     mbox->comm_queue = new std::deque<smx_synchro_t>();
45     mbox->done_comm_queue = nullptr; // Allocated on need only
46     mbox->permanent_receiver=NULL;
47
48     XBT_DEBUG("Creating a mailbox at %p with name %s", mbox, name);
49     xbt_dict_set(mailboxes, name, mbox, NULL);
50   }
51   return mbox;
52 }
53
54 void SIMIX_mbox_free(void *data)
55 {
56   XBT_DEBUG("mbox free %p", data);
57   smx_mailbox_t mbox = (smx_mailbox_t) data;
58   delete mbox->comm_queue;
59   delete mbox->done_comm_queue;
60
61   xbt_free(mbox);
62 }
63
64 smx_mailbox_t SIMIX_mbox_get_by_name(const char *name)
65 {
66   return (smx_mailbox_t) xbt_dict_get_or_null(mailboxes, name);
67 }
68
69 smx_synchro_t SIMIX_mbox_get_head(smx_mailbox_t mbox)
70 {
71   return mbox->comm_queue->empty()? nullptr:mbox->comm_queue->front();
72 }
73
74 /**
75  *  \brief get the receiver (process associated to the mailbox)
76  *  \param mbox The rendez-vous point
77  *  \return process The receiving process (NULL if not set)
78  */
79 smx_process_t SIMIX_mbox_get_receiver(smx_mailbox_t mbox)
80 {
81   return mbox->permanent_receiver;
82 }
83
84 /**
85  *  \brief set the receiver of the rendez vous point to allow eager sends
86  *  \param mbox The rendez-vous point
87  *  \param process The receiving process
88  */
89 void SIMIX_mbox_set_receiver(smx_mailbox_t mbox, smx_process_t process)
90 {
91   mbox->permanent_receiver=process;
92   if (mbox->done_comm_queue == nullptr)
93     mbox->done_comm_queue = new std::deque<smx_synchro_t>();
94 }
95
96 /**
97  *  \brief Pushes a communication synchro into a rendez-vous point
98  *  \param mbox The mailbox
99  *  \param comm The communication synchro
100  */
101 static inline void SIMIX_mbox_push(smx_mailbox_t mbox, smx_synchro_t comm)
102 {
103   mbox->comm_queue->push_back(comm);
104   comm->comm.mbox = mbox;
105 }
106
107 /**
108  *  \brief Removes a communication synchro from a rendez-vous point
109  *  \param mbox The rendez-vous point
110  *  \param comm The communication synchro
111  */
112 void SIMIX_mbox_remove(smx_mailbox_t mbox, smx_synchro_t comm)
113 {
114   comm->comm.mbox = NULL;
115   for (auto it = mbox->comm_queue->begin(); it != mbox->comm_queue->end(); it++)
116     if (*it == comm) {
117       mbox->comm_queue->erase(it);
118       return;
119     }
120   xbt_die("Cannot remove this comm that is not part of the mailbox");
121 }
122
123 /**
124  *  \brief Checks if there is a communication synchro queued in a deque matching our needs
125  *  \param type The type of communication we are looking for (comm_send, comm_recv)
126  *  \return The communication synchro if found, NULL otherwise
127  */
128 static smx_synchro_t _find_matching_comm(std::deque<smx_synchro_t> *deque, e_smx_comm_type_t type,
129     int (*match_fun)(void *, void *,smx_synchro_t), void *this_user_data, smx_synchro_t my_synchro, bool remove_matching)
130 {
131   void* other_user_data = NULL;
132
133   for(auto it = deque->begin(); it != deque->end(); it++){
134     smx_synchro_t synchro = *it;
135     if (synchro->comm.type == SIMIX_COMM_SEND) {
136       other_user_data = synchro->comm.src_data;
137     } else if (synchro->comm.type == SIMIX_COMM_RECEIVE) {
138       other_user_data = synchro->comm.dst_data;
139     }
140     if (synchro->comm.type == type &&
141         (!match_fun               ||               match_fun(this_user_data,  other_user_data, synchro)) &&
142         (!synchro->comm.match_fun || synchro->comm.match_fun(other_user_data, this_user_data,  my_synchro))) {
143       XBT_DEBUG("Found a matching communication synchro %p", synchro);
144       if (remove_matching)
145         deque->erase(it);
146       synchro->comm.refcount++;
147 #if HAVE_MC
148       synchro->comm.mbox_cpy = synchro->comm.mbox;
149 #endif
150       synchro->comm.mbox = NULL;
151       return synchro;
152     }
153     XBT_DEBUG("Sorry, communication synchro %p does not match our needs:"
154               " its type is %d but we are looking for a comm of type %d (or maybe the filtering didn't match)",
155               synchro, (int)synchro->comm.type, (int)type);
156   }
157   XBT_DEBUG("No matching communication synchro found");
158   return NULL;
159 }
160
161 /******************************************************************************/
162 /*                          Communication synchros                            */
163 /******************************************************************************/
164
165 /**
166  *  \brief Creates a new communicate synchro
167  *  \param type The direction of communication (comm_send, comm_recv)
168  *  \return The new communicate synchro
169  */
170 smx_synchro_t SIMIX_comm_new(e_smx_comm_type_t type)
171 {
172   smx_synchro_t synchro;
173
174   /* alloc structures */
175   synchro = (smx_synchro_t) xbt_mallocator_get(simix_global->synchro_mallocator);
176
177   synchro->type = SIMIX_SYNC_COMMUNICATE;
178   synchro->state = SIMIX_WAITING;
179
180   /* set communication */
181   synchro->comm.type = type;
182   synchro->comm.refcount = 1;
183   synchro->comm.src_data=NULL;
184   synchro->comm.dst_data=NULL;
185
186   synchro->category = NULL;
187
188   XBT_DEBUG("Create communicate synchro %p", synchro);
189
190   return synchro;
191 }
192
193 /**
194  *  \brief Destroy a communicate synchro
195  *  \param synchro The communicate synchro to be destroyed
196  */
197 void SIMIX_comm_destroy(smx_synchro_t synchro)
198 {
199   XBT_DEBUG("Destroy synchro %p (refcount: %d), state: %d",
200             synchro, synchro->comm.refcount, (int)synchro->state);
201
202   if (synchro->comm.refcount <= 0) {
203     xbt_backtrace_display_current();
204     xbt_die("The refcount of comm %p is already 0 before decreasing it. "
205             "That's a bug! If you didn't test and/or wait the same communication twice in your code, then the bug is SimGrid's...", synchro);
206   }
207   synchro->comm.refcount--;
208   if (synchro->comm.refcount > 0)
209       return;
210   XBT_DEBUG("Really free communication %p; refcount is now %d", synchro,
211       synchro->comm.refcount);
212
213   xbt_free(synchro->name);
214   SIMIX_comm_destroy_internal_actions(synchro);
215
216   if (synchro->comm.detached && synchro->state != SIMIX_DONE) {
217     /* the communication has failed and was detached:
218      * we have to free the buffer */
219     if (synchro->comm.clean_fun) {
220       synchro->comm.clean_fun(synchro->comm.src_buff);
221     }
222     synchro->comm.src_buff = NULL;
223   }
224
225   if(synchro->comm.mbox)
226     SIMIX_mbox_remove(synchro->comm.mbox, synchro);
227
228   xbt_mallocator_release(simix_global->synchro_mallocator, synchro);
229 }
230
231 void SIMIX_comm_destroy_internal_actions(smx_synchro_t synchro)
232 {
233   if (synchro->comm.surf_comm){
234     synchro->comm.surf_comm->unref();
235     synchro->comm.surf_comm = NULL;
236   }
237
238   if (synchro->comm.src_timeout){
239     synchro->comm.src_timeout->unref();
240     synchro->comm.src_timeout = NULL;
241   }
242
243   if (synchro->comm.dst_timeout){
244     synchro->comm.dst_timeout->unref();
245     synchro->comm.dst_timeout = NULL;
246   }
247 }
248
249 void simcall_HANDLER_comm_send(smx_simcall_t simcall, smx_process_t src, smx_mailbox_t mbox,
250                                   double task_size, double rate,
251                                   void *src_buff, size_t src_buff_size,
252                                   int (*match_fun)(void *, void *,smx_synchro_t),
253                                   void (*copy_data_fun)(smx_synchro_t, void*, size_t),
254           void *data, double timeout){
255   smx_synchro_t comm = simcall_HANDLER_comm_isend(simcall, src, mbox, task_size, rate,
256                            src_buff, src_buff_size, match_fun, NULL, copy_data_fun,
257                data, 0);
258   SIMCALL_SET_MC_VALUE(simcall, 0);
259   simcall_HANDLER_comm_wait(simcall, comm, timeout);
260 }
261 smx_synchro_t simcall_HANDLER_comm_isend(smx_simcall_t simcall, smx_process_t src_proc, smx_mailbox_t mbox,
262                                   double task_size, double rate,
263                                   void *src_buff, size_t src_buff_size,
264                                   int (*match_fun)(void *, void *,smx_synchro_t),
265                                   void (*clean_fun)(void *), // used to free the synchro in case of problem after a detached send
266                                   void (*copy_data_fun)(smx_synchro_t, void*, size_t),// used to copy data if not default one
267                           void *data, int detached)
268 {
269   XBT_DEBUG("send from %p", mbox);
270
271   /* Prepare a synchro describing us, so that it gets passed to the user-provided filter of other side */
272   smx_synchro_t this_synchro = SIMIX_comm_new(SIMIX_COMM_SEND);
273
274   /* Look for communication synchro matching our needs. We also provide a description of
275    * ourself so that the other side also gets a chance of choosing if it wants to match with us.
276    *
277    * If it is not found then push our communication into the rendez-vous point */
278   smx_synchro_t other_synchro =
279       _find_matching_comm(mbox->comm_queue, SIMIX_COMM_RECEIVE, match_fun, data, this_synchro, /*remove_matching*/true);
280
281   if (!other_synchro) {
282     other_synchro = this_synchro;
283
284     if (mbox->permanent_receiver!=NULL){
285       //this mailbox is for small messages, which have to be sent right now
286       other_synchro->state = SIMIX_READY;
287       other_synchro->comm.dst_proc=mbox->permanent_receiver;
288       other_synchro->comm.refcount++;
289       mbox->done_comm_queue->push_back(other_synchro);
290       other_synchro->comm.mbox=mbox;
291       XBT_DEBUG("pushing a message into the permanent receive fifo %p, comm %p", mbox, &(other_synchro->comm));
292
293     }else{
294       SIMIX_mbox_push(mbox, this_synchro);
295     }
296   } else {
297     XBT_DEBUG("Receive already pushed");
298
299     SIMIX_comm_destroy(this_synchro);
300
301     other_synchro->state = SIMIX_READY;
302     other_synchro->comm.type = SIMIX_COMM_READY;
303
304   }
305   xbt_fifo_push(src_proc->comms, other_synchro);
306
307   /* if the communication synchro is detached then decrease the refcount
308    * by one, so it will be eliminated by the receiver's destroy call */
309   if (detached) {
310     other_synchro->comm.detached = 1;
311     other_synchro->comm.refcount--;
312     other_synchro->comm.clean_fun = clean_fun;
313   } else {
314     other_synchro->comm.clean_fun = NULL;
315   }
316
317   /* Setup the communication synchro */
318   other_synchro->comm.src_proc = src_proc;
319   other_synchro->comm.task_size = task_size;
320   other_synchro->comm.rate = rate;
321   other_synchro->comm.src_buff = src_buff;
322   other_synchro->comm.src_buff_size = src_buff_size;
323   other_synchro->comm.src_data = data;
324
325   other_synchro->comm.match_fun = match_fun;
326   other_synchro->comm.copy_data_fun = copy_data_fun;
327
328
329   if (MC_is_active() || MC_record_replay_is_active()) {
330     other_synchro->state = SIMIX_RUNNING;
331     return (detached ? NULL : other_synchro);
332   }
333
334   SIMIX_comm_start(other_synchro);
335   return (detached ? NULL : other_synchro);
336 }
337
338 void simcall_HANDLER_comm_recv(smx_simcall_t simcall, smx_process_t receiver, smx_mailbox_t mbox,
339                          void *dst_buff, size_t *dst_buff_size,
340                          int (*match_fun)(void *, void *, smx_synchro_t),
341                          void (*copy_data_fun)(smx_synchro_t, void*, size_t),
342                          void *data, double timeout, double rate)
343 {
344   smx_synchro_t comm = SIMIX_comm_irecv(receiver, mbox, dst_buff,
345                            dst_buff_size, match_fun, copy_data_fun, data, rate);
346   SIMCALL_SET_MC_VALUE(simcall, 0);
347   simcall_HANDLER_comm_wait(simcall, comm, timeout);
348 }
349
350 smx_synchro_t simcall_HANDLER_comm_irecv(smx_simcall_t simcall, smx_process_t receiver, smx_mailbox_t mbox,
351     void *dst_buff, size_t *dst_buff_size,
352     int (*match_fun)(void *, void *, smx_synchro_t),
353     void (*copy_data_fun)(smx_synchro_t, void*, size_t),
354     void *data, double rate)
355 {
356   return SIMIX_comm_irecv(receiver, mbox, dst_buff, dst_buff_size, match_fun, copy_data_fun, data, rate);
357 }
358
359 smx_synchro_t SIMIX_comm_irecv(smx_process_t dst_proc, smx_mailbox_t mbox, void *dst_buff, size_t *dst_buff_size,
360     int (*match_fun)(void *, void *, smx_synchro_t),
361     void (*copy_data_fun)(smx_synchro_t, void*, size_t), // used to copy data if not default one
362     void *data, double rate)
363 {
364   XBT_DEBUG("recv from %p %p", mbox, mbox->comm_queue);
365   smx_synchro_t this_synchro = SIMIX_comm_new(SIMIX_COMM_RECEIVE);
366
367   smx_synchro_t other_synchro;
368   //communication already done, get it inside the fifo of completed comms
369   if (mbox->permanent_receiver && ! mbox->done_comm_queue->empty()) {
370
371     XBT_DEBUG("We have a comm that has probably already been received, trying to match it, to skip the communication");
372     //find a match in the already received fifo
373     other_synchro = _find_matching_comm(mbox->done_comm_queue, SIMIX_COMM_SEND, match_fun, data, this_synchro,/*remove_matching*/true);
374     //if not found, assume the receiver came first, register it to the mailbox in the classical way
375     if (!other_synchro)  {
376       XBT_DEBUG("We have messages in the permanent receive list, but not the one we are looking for, pushing request into fifo");
377       other_synchro = this_synchro;
378       SIMIX_mbox_push(mbox, this_synchro);
379     } else {
380       if(other_synchro->comm.surf_comm && SIMIX_comm_get_remains(other_synchro)==0.0) {
381         XBT_DEBUG("comm %p has been already sent, and is finished, destroy it",&(other_synchro->comm));
382         other_synchro->state = SIMIX_DONE;
383         other_synchro->comm.type = SIMIX_COMM_DONE;
384         other_synchro->comm.mbox = NULL;
385       }
386       other_synchro->comm.refcount--;
387       SIMIX_comm_destroy(this_synchro);
388     }
389   } else {
390     /* Prepare a synchro describing us, so that it gets passed to the user-provided filter of other side */
391
392     /* Look for communication synchro matching our needs. We also provide a description of
393      * ourself so that the other side also gets a chance of choosing if it wants to match with us.
394      *
395      * If it is not found then push our communication into the rendez-vous point */
396     other_synchro = _find_matching_comm(mbox->comm_queue, SIMIX_COMM_SEND, match_fun, data, this_synchro,/*remove_matching*/true);
397
398     if (!other_synchro) {
399       XBT_DEBUG("Receive pushed first %zu", mbox->comm_queue->size());
400       other_synchro = this_synchro;
401       SIMIX_mbox_push(mbox, this_synchro);
402     } else {
403       SIMIX_comm_destroy(this_synchro);
404       other_synchro->state = SIMIX_READY;
405       other_synchro->comm.type = SIMIX_COMM_READY;
406       //other_synchro->comm.refcount--;
407     }
408     xbt_fifo_push(dst_proc->comms, other_synchro);
409   }
410
411   /* Setup communication synchro */
412   other_synchro->comm.dst_proc = dst_proc;
413   other_synchro->comm.dst_buff = dst_buff;
414   other_synchro->comm.dst_buff_size = dst_buff_size;
415   other_synchro->comm.dst_data = data;
416
417   if (rate != -1.0 && (other_synchro->comm.rate == -1.0 || rate < other_synchro->comm.rate))
418     other_synchro->comm.rate = rate;
419
420   other_synchro->comm.match_fun = match_fun;
421   other_synchro->comm.copy_data_fun = copy_data_fun;
422
423   if (MC_is_active() || MC_record_replay_is_active()) {
424     other_synchro->state = SIMIX_RUNNING;
425     return other_synchro;
426   }
427
428   SIMIX_comm_start(other_synchro);
429   return other_synchro;
430 }
431
432 smx_synchro_t simcall_HANDLER_comm_iprobe(smx_simcall_t simcall, smx_mailbox_t mbox,
433                                    int type, int src, int tag,
434                                    int (*match_fun)(void *, void *, smx_synchro_t),
435                                    void *data){
436   return SIMIX_comm_iprobe(simcall->issuer, mbox, type, src, tag, match_fun, data);
437 }
438
439 smx_synchro_t SIMIX_comm_iprobe(smx_process_t dst_proc, smx_mailbox_t mbox, int type, int src,
440                               int tag, int (*match_fun)(void *, void *, smx_synchro_t), void *data)
441 {
442   XBT_DEBUG("iprobe from %p %p", mbox, mbox->comm_queue);
443   smx_synchro_t this_synchro;
444   int smx_type;
445   if(type == 1){
446     this_synchro=SIMIX_comm_new(SIMIX_COMM_SEND);
447     smx_type = SIMIX_COMM_RECEIVE;
448   } else{
449     this_synchro=SIMIX_comm_new(SIMIX_COMM_RECEIVE);
450     smx_type = SIMIX_COMM_SEND;
451   } 
452   smx_synchro_t other_synchro=NULL;
453   if(mbox->permanent_receiver && ! mbox->done_comm_queue->empty()){
454     XBT_DEBUG("first check in the permanent recv mailbox, to see if we already got something");
455     other_synchro =
456         _find_matching_comm(mbox->done_comm_queue, (e_smx_comm_type_t) smx_type, match_fun, data, this_synchro,/*remove_matching*/false);
457   }
458   if (!other_synchro){
459     XBT_DEBUG("check if we have more luck in the normal mailbox");
460     other_synchro =
461         _find_matching_comm(mbox->comm_queue, (e_smx_comm_type_t) smx_type, match_fun, data, this_synchro,/*remove_matching*/false);
462   }
463   if(other_synchro)
464     other_synchro->comm.refcount--;
465
466   SIMIX_comm_destroy(this_synchro);
467   return other_synchro;
468 }
469
470 void simcall_HANDLER_comm_wait(smx_simcall_t simcall, smx_synchro_t synchro, double timeout)
471 {
472   /* the simcall may be a wait, a send or a recv */
473   surf_action_t sleep;
474
475   /* Associate this simcall to the wait synchro */
476   XBT_DEBUG("simcall_HANDLER_comm_wait, %p", synchro);
477
478   xbt_fifo_push(synchro->simcalls, simcall);
479   simcall->issuer->waiting_synchro = synchro;
480
481   if (MC_is_active() || MC_record_replay_is_active()) {
482     int idx = SIMCALL_GET_MC_VALUE(simcall);
483     if (idx == 0) {
484       synchro->state = SIMIX_DONE;
485     } else {
486       /* If we reached this point, the wait simcall must have a timeout */
487       /* Otherwise it shouldn't be enabled and executed by the MC */
488       if (timeout == -1)
489         THROW_IMPOSSIBLE;
490
491       if (synchro->comm.src_proc == simcall->issuer)
492         synchro->state = SIMIX_SRC_TIMEOUT;
493       else
494         synchro->state = SIMIX_DST_TIMEOUT;
495     }
496
497     SIMIX_comm_finish(synchro);
498     return;
499   }
500
501   /* If the synchro has already finish perform the error handling, */
502   /* otherwise set up a waiting timeout on the right side          */
503   if (synchro->state != SIMIX_WAITING && synchro->state != SIMIX_RUNNING) {
504     SIMIX_comm_finish(synchro);
505   } 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 */
506     sleep = surf_host_sleep(simcall->issuer->host, timeout);
507     sleep->setData(synchro);
508
509     if (simcall->issuer == synchro->comm.src_proc)
510       synchro->comm.src_timeout = sleep;
511     else
512       synchro->comm.dst_timeout = sleep;
513   }
514 }
515
516 void simcall_HANDLER_comm_test(smx_simcall_t simcall, smx_synchro_t synchro)
517 {
518   if(MC_is_active() || MC_record_replay_is_active()){
519     simcall_comm_test__set__result(simcall, synchro->comm.src_proc && synchro->comm.dst_proc);
520     if(simcall_comm_test__get__result(simcall)){
521       synchro->state = SIMIX_DONE;
522       xbt_fifo_push(synchro->simcalls, simcall);
523       SIMIX_comm_finish(synchro);
524     }else{
525       SIMIX_simcall_answer(simcall);
526     }
527     return;
528   }
529
530   simcall_comm_test__set__result(simcall, (synchro->state != SIMIX_WAITING && synchro->state != SIMIX_RUNNING));
531   if (simcall_comm_test__get__result(simcall)) {
532     xbt_fifo_push(synchro->simcalls, simcall);
533     SIMIX_comm_finish(synchro);
534   } else {
535     SIMIX_simcall_answer(simcall);
536   }
537 }
538
539 void simcall_HANDLER_comm_testany(smx_simcall_t simcall, xbt_dynar_t synchros)
540 {
541   unsigned int cursor;
542   smx_synchro_t synchro;
543   simcall_comm_testany__set__result(simcall, -1);
544
545   if (MC_is_active() || MC_record_replay_is_active()){
546     int idx = SIMCALL_GET_MC_VALUE(simcall);
547     if(idx == -1){
548       SIMIX_simcall_answer(simcall);
549     }else{
550       synchro = xbt_dynar_get_as(synchros, idx, smx_synchro_t);
551       simcall_comm_testany__set__result(simcall, idx);
552       xbt_fifo_push(synchro->simcalls, simcall);
553       synchro->state = SIMIX_DONE;
554       SIMIX_comm_finish(synchro);
555     }
556     return;
557   }
558
559   xbt_dynar_foreach(simcall_comm_testany__get__comms(simcall), cursor,synchro) {
560     if (synchro->state != SIMIX_WAITING && synchro->state != SIMIX_RUNNING) {
561       simcall_comm_testany__set__result(simcall, cursor);
562       xbt_fifo_push(synchro->simcalls, simcall);
563       SIMIX_comm_finish(synchro);
564       return;
565     }
566   }
567   SIMIX_simcall_answer(simcall);
568 }
569
570 void simcall_HANDLER_comm_waitany(smx_simcall_t simcall, xbt_dynar_t synchros)
571 {
572   smx_synchro_t synchro;
573   unsigned int cursor = 0;
574
575   if (MC_is_active() || MC_record_replay_is_active()){
576     int idx = SIMCALL_GET_MC_VALUE(simcall);
577     synchro = xbt_dynar_get_as(synchros, idx, smx_synchro_t);
578     xbt_fifo_push(synchro->simcalls, simcall);
579     simcall_comm_waitany__set__result(simcall, idx);
580     synchro->state = SIMIX_DONE;
581     SIMIX_comm_finish(synchro);
582     return;
583   }
584
585   xbt_dynar_foreach(synchros, cursor, synchro){
586     /* associate this simcall to the the synchro */
587     xbt_fifo_push(synchro->simcalls, simcall);
588
589     /* see if the synchro is already finished */
590     if (synchro->state != SIMIX_WAITING && synchro->state != SIMIX_RUNNING){
591       SIMIX_comm_finish(synchro);
592       break;
593     }
594   }
595 }
596
597 void SIMIX_waitany_remove_simcall_from_actions(smx_simcall_t simcall)
598 {
599   smx_synchro_t synchro;
600   unsigned int cursor = 0;
601   xbt_dynar_t synchros = simcall_comm_waitany__get__comms(simcall);
602
603   xbt_dynar_foreach(synchros, cursor, synchro) {
604     xbt_fifo_remove(synchro->simcalls, simcall);
605   }
606 }
607
608 /**
609  *  \brief Starts the simulation of a communication synchro.
610  *  \param synchro the communication synchro
611  */
612 static inline void SIMIX_comm_start(smx_synchro_t synchro)
613 {
614   /* If both the sender and the receiver are already there, start the communication */
615   if (synchro->state == SIMIX_READY) {
616
617     sg_host_t sender = synchro->comm.src_proc->host;
618     sg_host_t receiver = synchro->comm.dst_proc->host;
619
620     XBT_DEBUG("Starting communication %p from '%s' to '%s'", synchro,
621               sg_host_get_name(sender), sg_host_get_name(receiver));
622
623     synchro->comm.surf_comm = surf_network_model_communicate(surf_network_model,
624                                                             sender, receiver,
625                                                             synchro->comm.task_size, synchro->comm.rate);
626
627     synchro->comm.surf_comm->setData(synchro);
628
629     synchro->state = SIMIX_RUNNING;
630
631     /* If a link is failed, detect it immediately */
632     if (synchro->comm.surf_comm->getState() == simgrid::surf::Action::State::failed) {
633       XBT_DEBUG("Communication from '%s' to '%s' failed to start because of a link failure",
634                 sg_host_get_name(sender), sg_host_get_name(receiver));
635       synchro->state = SIMIX_LINK_FAILURE;
636       SIMIX_comm_destroy_internal_actions(synchro);
637     }
638
639     /* If any of the process is suspend, create the synchro but stop its execution,
640        it will be restarted when the sender process resume */
641     if (SIMIX_process_is_suspended(synchro->comm.src_proc) ||
642         SIMIX_process_is_suspended(synchro->comm.dst_proc)) {
643       /* FIXME: check what should happen with the synchro state */
644
645       if (SIMIX_process_is_suspended(synchro->comm.src_proc))
646         XBT_DEBUG("The communication is suspended on startup because src (%s:%s) were suspended since it initiated the communication",
647                   sg_host_get_name(synchro->comm.src_proc->host), synchro->comm.src_proc->name);
648       else
649         XBT_DEBUG("The communication is suspended on startup because dst (%s:%s) were suspended since it initiated the communication",
650                   sg_host_get_name(synchro->comm.dst_proc->host), synchro->comm.dst_proc->name);
651
652       synchro->comm.surf_comm->suspend();
653
654     }
655   }
656 }
657
658 /**
659  * \brief Answers the SIMIX simcalls associated to a communication synchro.
660  * \param synchro a finished communication synchro
661  */
662 void SIMIX_comm_finish(smx_synchro_t synchro)
663 {
664   unsigned int destroy_count = 0;
665   smx_simcall_t simcall;
666
667   while ((simcall = (smx_simcall_t) xbt_fifo_shift(synchro->simcalls))) {
668
669     /* If a waitany simcall is waiting for this synchro to finish, then remove
670        it from the other synchros in the waitany list. Afterwards, get the
671        position of the actual synchro in the waitany dynar and
672        return it as the result of the simcall */
673
674     if (simcall->call == SIMCALL_NONE) //FIXME: maybe a better way to handle this case
675       continue; // if process handling comm is killed
676     if (simcall->call == SIMCALL_COMM_WAITANY) {
677       SIMIX_waitany_remove_simcall_from_actions(simcall);
678       if (!MC_is_active() && !MC_record_replay_is_active())
679         simcall_comm_waitany__set__result(simcall, xbt_dynar_search(simcall_comm_waitany__get__comms(simcall), &synchro));
680     }
681
682     /* If the synchro is still in a rendez-vous point then remove from it */
683     if (synchro->comm.mbox)
684       SIMIX_mbox_remove(synchro->comm.mbox, synchro);
685
686     XBT_DEBUG("SIMIX_comm_finish: synchro state = %d", (int)synchro->state);
687
688     /* Check out for errors */
689
690     if (simcall->issuer->host->isOff()) {
691       simcall->issuer->context->iwannadie = 1;
692       SMX_EXCEPTION(simcall->issuer, host_error, 0, "Host failed");
693     } else
694
695     switch (synchro->state) {
696
697     case SIMIX_DONE:
698       XBT_DEBUG("Communication %p complete!", synchro);
699       SIMIX_comm_copy_data(synchro);
700       break;
701
702     case SIMIX_SRC_TIMEOUT:
703       SMX_EXCEPTION(simcall->issuer, timeout_error, 0,
704                     "Communication timeouted because of sender");
705       break;
706
707     case SIMIX_DST_TIMEOUT:
708       SMX_EXCEPTION(simcall->issuer, timeout_error, 0,
709                     "Communication timeouted because of receiver");
710       break;
711
712     case SIMIX_SRC_HOST_FAILURE:
713       if (simcall->issuer == synchro->comm.src_proc)
714         simcall->issuer->context->iwannadie = 1;
715 //          SMX_EXCEPTION(simcall->issuer, host_error, 0, "Host failed");
716       else
717         SMX_EXCEPTION(simcall->issuer, network_error, 0, "Remote peer failed");
718       break;
719
720     case SIMIX_DST_HOST_FAILURE:
721       if (simcall->issuer == synchro->comm.dst_proc)
722         simcall->issuer->context->iwannadie = 1;
723 //          SMX_EXCEPTION(simcall->issuer, host_error, 0, "Host failed");
724       else
725         SMX_EXCEPTION(simcall->issuer, network_error, 0, "Remote peer failed");
726       break;
727
728     case SIMIX_LINK_FAILURE:
729
730       XBT_DEBUG("Link failure in synchro %p between '%s' and '%s': posting an exception to the issuer: %s (%p) detached:%d",
731                 synchro,
732                 synchro->comm.src_proc ? sg_host_get_name(synchro->comm.src_proc->host) : NULL,
733                 synchro->comm.dst_proc ? sg_host_get_name(synchro->comm.dst_proc->host) : NULL,
734                 simcall->issuer->name, simcall->issuer, synchro->comm.detached);
735       if (synchro->comm.src_proc == simcall->issuer) {
736         XBT_DEBUG("I'm source");
737       } else if (synchro->comm.dst_proc == simcall->issuer) {
738         XBT_DEBUG("I'm dest");
739       } else {
740         XBT_DEBUG("I'm neither source nor dest");
741       }
742       SMX_EXCEPTION(simcall->issuer, network_error, 0, "Link failure");
743       break;
744
745     case SIMIX_CANCELED:
746       if (simcall->issuer == synchro->comm.dst_proc)
747         SMX_EXCEPTION(simcall->issuer, cancel_error, 0,
748                       "Communication canceled by the sender");
749       else
750         SMX_EXCEPTION(simcall->issuer, cancel_error, 0,
751                       "Communication canceled by the receiver");
752       break;
753
754     default:
755       xbt_die("Unexpected synchro state in SIMIX_comm_finish: %d", (int)synchro->state);
756     }
757
758     /* if there is an exception during a waitany or a testany, indicate the position of the failed communication */
759     if (simcall->issuer->doexception) {
760       if (simcall->call == SIMCALL_COMM_WAITANY) {
761         simcall->issuer->running_ctx->exception.value = xbt_dynar_search(simcall_comm_waitany__get__comms(simcall), &synchro);
762       }
763       else if (simcall->call == SIMCALL_COMM_TESTANY) {
764         simcall->issuer->running_ctx->exception.value = xbt_dynar_search(simcall_comm_testany__get__comms(simcall), &synchro);
765       }
766     }
767
768     if (simcall->issuer->host->isOff()) {
769       simcall->issuer->context->iwannadie = 1;
770     }
771
772     simcall->issuer->waiting_synchro = NULL;
773     xbt_fifo_remove(simcall->issuer->comms, synchro);
774     if(synchro->comm.detached){
775       if(simcall->issuer == synchro->comm.src_proc){
776         if(synchro->comm.dst_proc)
777           xbt_fifo_remove(synchro->comm.dst_proc->comms, synchro);
778       }
779       if(simcall->issuer == synchro->comm.dst_proc){
780         if(synchro->comm.src_proc)
781           xbt_fifo_remove(synchro->comm.src_proc->comms, synchro);
782       }
783     }
784     SIMIX_simcall_answer(simcall);
785     destroy_count++;
786   }
787
788   while (destroy_count-- > 0)
789     SIMIX_comm_destroy(synchro);
790 }
791
792 /**
793  * \brief This function is called when a Surf communication synchro is finished.
794  * \param synchro the corresponding Simix communication
795  */
796 void SIMIX_post_comm(smx_synchro_t synchro)
797 {
798   /* Update synchro state */
799   if (synchro->comm.src_timeout &&
800       synchro->comm.src_timeout->getState() == simgrid::surf::Action::State::done)
801     synchro->state = SIMIX_SRC_TIMEOUT;
802   else if (synchro->comm.dst_timeout &&
803     synchro->comm.dst_timeout->getState() == simgrid::surf::Action::State::done)
804     synchro->state = SIMIX_DST_TIMEOUT;
805   else if (synchro->comm.src_timeout &&
806     synchro->comm.src_timeout->getState() == simgrid::surf::Action::State::failed)
807     synchro->state = SIMIX_SRC_HOST_FAILURE;
808   else if (synchro->comm.dst_timeout &&
809       synchro->comm.dst_timeout->getState() == simgrid::surf::Action::State::failed)
810     synchro->state = SIMIX_DST_HOST_FAILURE;
811   else if (synchro->comm.surf_comm &&
812     synchro->comm.surf_comm->getState() == simgrid::surf::Action::State::failed) {
813     XBT_DEBUG("Puta madre. Surf says that the link broke");
814     synchro->state = SIMIX_LINK_FAILURE;
815   } else
816     synchro->state = SIMIX_DONE;
817
818   XBT_DEBUG("SIMIX_post_comm: comm %p, state %d, src_proc %p, dst_proc %p, detached: %d",
819             synchro, (int)synchro->state, synchro->comm.src_proc, synchro->comm.dst_proc, synchro->comm.detached);
820
821   /* destroy the surf actions associated with the Simix communication */
822   SIMIX_comm_destroy_internal_actions(synchro);
823
824   /* if there are simcalls associated with the synchro, then answer them */
825   if (xbt_fifo_size(synchro->simcalls)) {
826     SIMIX_comm_finish(synchro);
827   }
828 }
829
830 void SIMIX_comm_cancel(smx_synchro_t synchro)
831 {
832   /* if the synchro is a waiting state means that it is still in a mbox */
833   /* so remove from it and delete it */
834   if (synchro->state == SIMIX_WAITING) {
835     SIMIX_mbox_remove(synchro->comm.mbox, synchro);
836     synchro->state = SIMIX_CANCELED;
837   }
838   else if (!MC_is_active() /* when running the MC there are no surf actions */
839            && !MC_record_replay_is_active()
840            && (synchro->state == SIMIX_READY || synchro->state == SIMIX_RUNNING)) {
841
842     synchro->comm.surf_comm->cancel();
843   }
844 }
845
846 void SIMIX_comm_suspend(smx_synchro_t synchro)
847 {
848   /*FIXME: shall we suspend also the timeout synchro? */
849   if (synchro->comm.surf_comm)
850     synchro->comm.surf_comm->suspend();
851   /* in the other case, the action will be suspended on creation, in SIMIX_comm_start() */
852 }
853
854 void SIMIX_comm_resume(smx_synchro_t synchro)
855 {
856   /*FIXME: check what happen with the timeouts */
857   if (synchro->comm.surf_comm)
858     synchro->comm.surf_comm->resume();
859   /* in the other case, the synchro were not really suspended yet, see SIMIX_comm_suspend() and SIMIX_comm_start() */
860 }
861
862
863 /************* synchro Getters **************/
864
865 /**
866  *  \brief get the amount remaining from the communication
867  *  \param synchro The communication
868  */
869 double SIMIX_comm_get_remains(smx_synchro_t synchro)
870 {
871   if(!synchro)
872     return 0;
873
874   double remains;
875   switch (synchro->state) {
876
877   case SIMIX_RUNNING:
878     remains = synchro->comm.surf_comm->getRemains();
879     break;
880
881   case SIMIX_WAITING:
882   case SIMIX_READY:
883     remains = 0; /*FIXME: check what should be returned */
884     break;
885
886   default:
887     remains = 0; /*FIXME: is this correct? */
888     break;
889   }
890   return remains;
891 }
892
893 e_smx_state_t SIMIX_comm_get_state(smx_synchro_t synchro)
894 {
895   return synchro->state;
896 }
897
898 /**
899  *  \brief Return the user data associated to the sender of the communication
900  *  \param synchro The communication
901  *  \return the user data
902  */
903 void* SIMIX_comm_get_src_data(smx_synchro_t synchro)
904 {
905   return synchro->comm.src_data;
906 }
907
908 /**
909  *  \brief Return the user data associated to the receiver of the communication
910  *  \param synchro The communication
911  *  \return the user data
912  */
913 void* SIMIX_comm_get_dst_data(smx_synchro_t synchro)
914 {
915   return synchro->comm.dst_data;
916 }
917
918 smx_process_t SIMIX_comm_get_src_proc(smx_synchro_t synchro)
919 {
920   return synchro->comm.src_proc;
921 }
922
923 smx_process_t SIMIX_comm_get_dst_proc(smx_synchro_t synchro)
924 {
925   return synchro->comm.dst_proc;
926 }
927
928 /******************************************************************************/
929 /*                    SIMIX_comm_copy_data callbacks                       */
930 /******************************************************************************/
931 static void (*SIMIX_comm_copy_data_callback) (smx_synchro_t, void*, size_t) =
932   &SIMIX_comm_copy_pointer_callback;
933
934 void
935 SIMIX_comm_set_copy_data_callback(void (*callback) (smx_synchro_t, void*, size_t))
936 {
937   SIMIX_comm_copy_data_callback = callback;
938 }
939
940 void SIMIX_comm_copy_pointer_callback(smx_synchro_t comm, void* buff, size_t buff_size)
941 {
942   xbt_assert((buff_size == sizeof(void *)), "Cannot copy %zu bytes: must be sizeof(void*)", buff_size);
943   *(void **) (comm->comm.dst_buff) = buff;
944 }
945
946 void SIMIX_comm_copy_buffer_callback(smx_synchro_t comm, void* buff, size_t buff_size)
947 {
948   XBT_DEBUG("Copy the data over");
949   memcpy(comm->comm.dst_buff, buff, buff_size);
950   if (comm->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
951     xbt_free(buff);
952     comm->comm.src_buff = NULL;
953   }
954 }
955
956
957 /**
958  *  \brief Copy the communication data from the sender's buffer to the receiver's one
959  *  \param comm The communication
960  */
961 void SIMIX_comm_copy_data(smx_synchro_t comm)
962 {
963   size_t buff_size = comm->comm.src_buff_size;
964   /* If there is no data to be copy then return */
965   if (!comm->comm.src_buff || !comm->comm.dst_buff || comm->comm.copied)
966     return;
967
968   XBT_DEBUG("Copying comm %p data from %s (%p) -> %s (%p) (%zu bytes)",
969             comm,
970             comm->comm.src_proc ? sg_host_get_name(comm->comm.src_proc->host) : "a finished process",
971             comm->comm.src_buff,
972             comm->comm.dst_proc ? sg_host_get_name(comm->comm.dst_proc->host) : "a finished process",
973             comm->comm.dst_buff, buff_size);
974
975   /* Copy at most dst_buff_size bytes of the message to receiver's buffer */
976   if (comm->comm.dst_buff_size)
977     buff_size = MIN(buff_size, *(comm->comm.dst_buff_size));
978
979   /* Update the receiver's buffer size to the copied amount */
980   if (comm->comm.dst_buff_size)
981     *comm->comm.dst_buff_size = buff_size;
982
983   if (buff_size > 0){
984       if(comm->comm.copy_data_fun)
985         comm->comm.copy_data_fun (comm, comm->comm.src_buff, buff_size);
986       else
987         SIMIX_comm_copy_data_callback (comm, comm->comm.src_buff, buff_size);
988   }
989
990
991   /* Set the copied flag so we copy data only once */
992   /* (this function might be called from both communication ends) */
993   comm->comm.copied = 1;
994 }