Logo AND Algorithmique Numérique Distribuée

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