Logo AND Algorithmique Numérique Distribuée

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