Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
9632b26a06521a7bc5874b5f20c1fbc16c36202d
[simgrid.git] / src / msg / msg_gos.cpp
1 /* Copyright (c) 2004-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/simix/smx_private.h" /* MSG_task_listen looks inside the rdv directly. Not clean. */
7 #include "msg_private.h"
8 #include "mc/mc.h"
9 #include "xbt/log.h"
10 #include "xbt/sysdep.h"
11
12 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(msg_gos, msg,
13                                 "Logging specific to MSG (gos)");
14
15 /** \ingroup msg_task_usage
16  * \brief Executes a task and waits for its termination.
17  *
18  * This function is used for describing the behavior of a process. It takes only one parameter.
19  * \param task a #msg_task_t to execute on the location on which the process is running.
20  * \return #MSG_OK if the task was successfully completed, #MSG_TASK_CANCELED or #MSG_HOST_FAILURE otherwise
21  */
22 msg_error_t MSG_task_execute(msg_task_t task)
23 {
24   /* TODO: add this to other locations */
25   msg_host_t host = MSG_process_get_host(MSG_process_self());
26   MSG_host_add_task(host, task);
27
28   msg_error_t ret = MSG_parallel_task_execute(task);
29
30   MSG_host_del_task(host, task);
31
32   return ret;
33 }
34
35 /** \ingroup msg_task_usage
36  * \brief Executes a parallel task and waits for its termination.
37  *
38  * \param task a #msg_task_t to execute on the location on which the process is running.
39  *
40  * \return #MSG_OK if the task was successfully completed, #MSG_TASK_CANCELED
41  * or #MSG_HOST_FAILURE otherwise
42  */
43 msg_error_t MSG_parallel_task_execute(msg_task_t task)
44 {
45   xbt_ex_t e;
46   simdata_task_t simdata = task->simdata;
47   simdata_process_t p_simdata = (simdata_process_t) SIMIX_process_self_get_data();
48   e_smx_state_t comp_state;
49   msg_error_t status = MSG_OK;
50
51   TRACE_msg_task_execute_start(task);
52
53   xbt_assert((!simdata->compute) && (task->simdata->isused == 0),
54              "This task is executed somewhere else. Go fix your code! %d", task->simdata->isused!=NULL);
55
56   XBT_DEBUG("Computing on %s", MSG_process_get_name(MSG_process_self()));
57
58   if (simdata->flops_amount == 0 && !simdata->host_nb) {
59     TRACE_msg_task_execute_end(task);
60     return MSG_OK;
61   }
62
63   TRY {
64     if (msg_global->debug_multiple_use)
65       MSG_BT(simdata->isused, "Using Backtrace");
66     else
67       simdata->isused = (void*)1;
68
69     if (simdata->host_nb > 0) {
70       simdata->compute = simcall_execution_parallel_start(task->name, simdata->host_nb,simdata->host_list,
71                                                        simdata->flops_parallel_amount, simdata->bytes_parallel_amount,
72                                                        1.0, -1.0);
73       XBT_DEBUG("Parallel execution action created: %p", simdata->compute);
74     } else {
75       unsigned long affinity_mask =
76          (unsigned long)(uintptr_t) xbt_dict_get_or_null_ext(simdata->affinity_mask_db, (char *) p_simdata->m_host,
77                                                              sizeof(msg_host_t));
78       XBT_DEBUG("execute %s@%s with affinity(0x%04lx)",
79                 MSG_task_get_name(task), MSG_host_get_name(p_simdata->m_host), affinity_mask);
80
81       simdata->compute = simcall_execution_start(task->name, simdata->flops_amount, simdata->priority,
82                                                  simdata->bound, affinity_mask);
83     }
84     simcall_set_category(simdata->compute, task->category);
85     p_simdata->waiting_action = simdata->compute;
86     comp_state = simcall_execution_wait(simdata->compute);
87
88     p_simdata->waiting_action = NULL;
89
90     if (msg_global->debug_multiple_use && simdata->isused!=0)
91       xbt_ex_free(*(xbt_ex_t*)simdata->isused);
92     simdata->isused = 0;
93
94     XBT_DEBUG("Execution task '%s' finished in state %d", task->name, (int)comp_state);
95   }
96   CATCH(e) {
97     switch (e.category) {
98     case cancel_error:
99       status = MSG_TASK_CANCELED;
100       break;
101     case host_error:
102       status = MSG_HOST_FAILURE;
103       break;
104     default:
105       RETHROW;
106     }
107     xbt_ex_free(e);
108   }
109   /* action ended, set comm and compute = NULL, the actions is already destroyed in the main function */
110   simdata->flops_amount = 0.0;
111   simdata->comm = NULL;
112   simdata->compute = NULL;
113   TRACE_msg_task_execute_end(task);
114
115   MSG_RETURN(status);
116 }
117
118 /** \ingroup msg_task_usage
119  * \brief Sleep for the specified number of seconds
120  *
121  * Makes the current process sleep until \a time seconds have elapsed.
122  *
123  * \param nb_sec a number of second
124  */
125 msg_error_t MSG_process_sleep(double nb_sec)
126 {
127   xbt_ex_t e;
128   msg_error_t status = MSG_OK;
129   /*msg_process_t proc = MSG_process_self();*/
130
131   TRACE_msg_process_sleep_in(MSG_process_self());
132
133   TRY {
134     simcall_process_sleep(nb_sec);
135   }
136   CATCH(e) {
137     switch (e.category) {
138     case cancel_error:
139       XBT_DEBUG("According to the JAVA API, a sleep call should only deal with HostFailureException, WTF here ?"); 
140       // adsein: MSG_TASK_CANCELED is assigned when someone kills the process that made the sleep, this is not
141       // correct. For instance, when the node is turned off, the error should be MSG_HOST_FAILURE, which is by the way
142       // and according to the JAVA document, the only exception that can be triggered by MSG_Process_sleep call.
143       // To avoid possible impacts in the code, I just raised a host_failure exception for the moment in the JAVA code
144       // and did not change anythings at the C level.
145       // See comment in the jmsg_process.c file, function JNIEXPORT void JNICALL Java_org_simgrid_msg_Process_sleep(JNIEnv *env, jclass cls, jlong jmillis, jint jnanos) 
146       status = MSG_TASK_CANCELED;
147       break;
148     default:
149       RETHROW;
150     }
151     xbt_ex_free(e);
152   }
153
154   TRACE_msg_process_sleep_out(MSG_process_self());
155   MSG_RETURN(status);
156 }
157
158 /** \ingroup msg_task_usage
159  * \brief Receives a task from a mailbox.
160  *
161  * This is a blocking function, the execution flow will be blocked until the task is received. See #MSG_task_irecv
162  * for receiving tasks asynchronously.
163  *
164  * \param task a memory location for storing a #msg_task_t.
165  * \param alias name of the mailbox to receive the task from
166  *
167  * \return Returns
168  * #MSG_OK if the task was successfully received,
169  * #MSG_HOST_FAILURE, or #MSG_TRANSFER_FAILURE otherwise.
170  */
171 msg_error_t MSG_task_receive(msg_task_t * task, const char *alias)
172 {
173   return MSG_task_receive_with_timeout(task, alias, -1);
174 }
175
176 /** \ingroup msg_task_usage
177  * \brief Receives a task from a mailbox at a given rate.
178  *
179  * \param task a memory location for storing a #msg_task_t.
180  * \param alias name of the mailbox to receive the task from
181  * \param rate limit the reception to rate bandwidth
182  *
183  * \return Returns
184  * #MSG_OK if the task was successfully received,
185  * #MSG_HOST_FAILURE, or #MSG_TRANSFER_FAILURE otherwise.
186  */
187 msg_error_t MSG_task_receive_bounded(msg_task_t * task, const char *alias, double rate)
188 {
189   return MSG_task_receive_with_timeout_bounded(task, alias, -1, rate);
190 }
191
192 /** \ingroup msg_task_usage
193  * \brief Receives a task from a mailbox with a given timeout.
194  *
195  * This is a blocking function with a timeout, the execution flow will be blocked until the task is received or the
196  * timeout is achieved. See #MSG_task_irecv for receiving tasks asynchronously.  You can provide a -1 timeout
197  * to obtain an infinite timeout.
198  *
199  * \param task a memory location for storing a #msg_task_t.
200  * \param alias name of the mailbox to receive the task from
201  * \param timeout is the maximum wait time for completion (if -1, this call is the same as #MSG_task_receive)
202  *
203  * \return Returns
204  * #MSG_OK if the task was successfully received,
205  * #MSG_HOST_FAILURE, or #MSG_TRANSFER_FAILURE, or #MSG_TIMEOUT otherwise.
206  */
207 msg_error_t MSG_task_receive_with_timeout(msg_task_t * task, const char *alias, double timeout)
208 {
209   return MSG_task_receive_ext(task, alias, timeout, NULL);
210 }
211
212 /** \ingroup msg_task_usage
213  * \brief Receives a task from a mailbox with a given timeout and at a given rate.
214  *
215  * \param task a memory location for storing a #msg_task_t.
216  * \param alias name of the mailbox to receive the task from
217  * \param timeout is the maximum wait time for completion (if -1, this call is the same as #MSG_task_receive)
218  *  \param rate limit the reception to rate bandwidth
219  *
220  * \return Returns
221  * #MSG_OK if the task was successfully received,
222  * #MSG_HOST_FAILURE, or #MSG_TRANSFER_FAILURE, or #MSG_TIMEOUT otherwise.
223  */
224 msg_error_t MSG_task_receive_with_timeout_bounded(msg_task_t * task, const char *alias, double timeout,double rate)
225 {
226   return MSG_task_receive_ext_bounded(task, alias, timeout, NULL, rate);
227 }
228
229 /** \ingroup msg_task_usage
230  * \brief Receives a task from a mailbox from a specific host with a given timeout.
231  *
232  * This is a blocking function with a timeout, the execution flow will be blocked until the task is received or the
233  * timeout is achieved. See #MSG_task_irecv for receiving tasks asynchronously. You can provide a -1 timeout
234  * to obtain an infinite timeout.
235  *
236  * \param task a memory location for storing a #msg_task_t.
237  * \param alias name of the mailbox to receive the task from
238  * \param timeout is the maximum wait time for completion (provide -1 for no timeout)
239  * \param host a #msg_host_t host from where the task was sent
240  *
241  * \return Returns
242  * #MSG_OK if the task was successfully received,
243 * #MSG_HOST_FAILURE, or #MSG_TRANSFER_FAILURE, or #MSG_TIMEOUT otherwise.
244  */
245 msg_error_t MSG_task_receive_ext(msg_task_t * task, const char *alias, double timeout, msg_host_t host)
246 {
247   xbt_ex_t e;
248   msg_error_t ret = MSG_OK;
249   XBT_DEBUG("MSG_task_receive_ext: Trying to receive a message on mailbox '%s'", alias);
250   TRY {
251     ret = MSG_mailbox_get_task_ext(MSG_mailbox_get_by_alias(alias), task, host, timeout);
252   }
253   CATCH(e) {
254     switch (e.category) {
255     case cancel_error:          /* may be thrown by MSG_mailbox_get_by_alias */
256       ret = MSG_HOST_FAILURE;
257       break;
258     default:
259       RETHROW;
260     }
261     xbt_ex_free(e);
262   }
263   return ret;
264 }
265
266 /** \ingroup msg_task_usage
267  * \brief Receives a task from a mailbox from a specific host with a given timeout  and at a given rate.
268  *
269  * \param task a memory location for storing a #msg_task_t.
270  * \param alias name of the mailbox to receive the task from
271  * \param timeout is the maximum wait time for completion (provide -1 for no timeout)
272  * \param host a #msg_host_t host from where the task was sent
273  * \param rate limit the reception to rate bandwidth
274  *
275  * \return Returns
276  * #MSG_OK if the task was successfully received,
277 * #MSG_HOST_FAILURE, or #MSG_TRANSFER_FAILURE, or #MSG_TIMEOUT otherwise.
278  */
279 msg_error_t MSG_task_receive_ext_bounded(msg_task_t * task, const char *alias, double timeout, msg_host_t host,
280                                          double rate)
281 {
282   XBT_DEBUG("MSG_task_receive_ext: Trying to receive a message on mailbox '%s'", alias);
283   return MSG_mailbox_get_task_ext_bounded(MSG_mailbox_get_by_alias(alias), task, host, timeout, rate);
284 }
285
286 /* Internal function used to factorize code between MSG_task_isend_with_matching() and MSG_task_dsend(). */
287 static inline msg_comm_t MSG_task_isend_internal(msg_task_t task, const char *alias,
288                                                      int (*match_fun)(void*,void*, smx_synchro_t),
289                                                      void *match_data, void_f_pvoid_t cleanup, int detached)
290 {
291   simdata_task_t t_simdata = NULL;
292   msg_process_t process = MSG_process_self();
293   msg_mailbox_t mailbox = MSG_mailbox_get_by_alias(alias);
294   int call_end = TRACE_msg_task_put_start(task);
295
296   /* Prepare the task to send */
297   t_simdata = task->simdata;
298   t_simdata->sender = process;
299   t_simdata->source = ((simdata_process_t) SIMIX_process_self_get_data())->m_host;
300
301   if (t_simdata->isused != 0) {
302     if (msg_global->debug_multiple_use){
303       XBT_ERROR("This task is already used in there:");
304       xbt_backtrace_display((xbt_ex_t*) t_simdata->isused);
305       XBT_ERROR("And you try to reuse it from here:");
306       xbt_backtrace_display_current();
307     } else {
308       xbt_assert(t_simdata->isused == 0,
309                  "This task is still being used somewhere else. You cannot send it now. Go fix your code!"
310                  "(use --cfg=msg/debug_multiple_use:on to get the backtrace of the other process)");
311     }
312   }
313
314   if (msg_global->debug_multiple_use)
315     MSG_BT(t_simdata->isused, "Using Backtrace");
316   else
317     t_simdata->isused = (void*)1;
318   t_simdata->comm = NULL;
319   msg_global->sent_msg++;
320
321   /* Send it by calling SIMIX network layer */
322   smx_synchro_t act = simcall_comm_isend(SIMIX_process_self(), mailbox, t_simdata->bytes_amount, t_simdata->rate,
323                                          task, sizeof(void *), match_fun, cleanup, NULL, match_data,detached);
324   t_simdata->comm = act; /* FIXME: is the field t_simdata->comm still useful? */
325
326   msg_comm_t comm;
327   if (detached) {
328     comm = NULL;
329   } else {
330     comm = xbt_new0(s_msg_comm_t, 1);
331     comm->task_sent = task;
332     comm->task_received = NULL;
333     comm->status = MSG_OK;
334     comm->s_comm = act;
335   }
336
337   if (TRACE_is_enabled())
338     simcall_set_category(act, task->category);
339   if (call_end)
340     TRACE_msg_task_put_end();
341
342   return comm;
343 }
344
345 /** \ingroup msg_task_usage
346  * \brief Sends a task on a mailbox.
347  *
348  * This is a non blocking function: use MSG_comm_wait() or MSG_comm_test() to end the communication.
349  *
350  * \param task a #msg_task_t to send on another location.
351  * \param alias name of the mailbox to sent the task to
352  * \return the msg_comm_t communication created
353  */
354 msg_comm_t MSG_task_isend(msg_task_t task, const char *alias)
355 {
356   return MSG_task_isend_internal(task, alias, NULL, NULL, NULL, 0);
357 }
358
359 /** \ingroup msg_task_usage
360  * \brief Sends a task on a mailbox with a maximum rate
361  *
362  * This is a non blocking function: use MSG_comm_wait() or MSG_comm_test() to end the communication. The maxrate
363  * parameter allows the application to limit the bandwidth utilization of network links when sending the task.
364  *
365  * \param task a #msg_task_t to send on another location.
366  * \param alias name of the mailbox to sent the task to
367  * \param maxrate the maximum communication rate for sending this task .
368  * \return the msg_comm_t communication created
369  */
370 msg_comm_t MSG_task_isend_bounded(msg_task_t task, const char *alias, double maxrate)
371 {
372   task->simdata->rate = maxrate;
373   return MSG_task_isend_internal(task, alias, NULL, NULL, NULL, 0);
374 }
375
376 /** \ingroup msg_task_usage
377  * \brief Sends a task on a mailbox, with support for matching requests
378  *
379  * This is a non blocking function: use MSG_comm_wait() or MSG_comm_test() to end the communication.
380  *
381  * \param task a #msg_task_t to send on another location.
382  * \param alias name of the mailbox to sent the task to
383  * \param match_fun boolean function which parameters are:
384  *        - match_data_provided_here
385  *        - match_data_provided_by_other_side_if_any
386  *        - the_smx_synchro_describing_the_other_side
387  * \param match_data user provided data passed to match_fun
388  * \return the msg_comm_t communication created
389  */
390 msg_comm_t MSG_task_isend_with_matching(msg_task_t task, const char *alias,
391                                         int (*match_fun)(void*, void*, smx_synchro_t), void *match_data)
392 {
393   return MSG_task_isend_internal(task, alias, match_fun, match_data, NULL, 0);
394 }
395
396 /** \ingroup msg_task_usage
397  * \brief Sends a task on a mailbox.
398  *
399  * This is a non blocking detached send function.
400  * Think of it as a best effort send. Keep in mind that the third parameter is only called if the communication fails.
401  * If the communication does work, it is responsibility of the receiver code to free anything related to the task, as
402  * usual. More details on this can be obtained on
403  * <a href="http://lists.gforge.inria.fr/pipermail/simgrid-user/2011-November/002649.html">this thread</a>
404  * in the SimGrid-user mailing list archive.
405  *
406  * \param task a #msg_task_t to send on another location.
407  * \param alias name of the mailbox to sent the task to
408  * \param cleanup a function to destroy the task if the communication fails, e.g. MSG_task_destroy
409  * (if NULL, no function will be called)
410  */
411 void MSG_task_dsend(msg_task_t task, const char *alias, void_f_pvoid_t cleanup)
412 {
413   MSG_task_isend_internal(task, alias, NULL, NULL, cleanup, 1);
414 }
415
416 /** \ingroup msg_task_usage
417  * \brief Sends a task on a mailbox with a maximal rate.
418  *
419  * This is a non blocking detached send function.
420  * Think of it as a best effort send. Keep in mind that the third parameter is only called if the communication fails.
421  * If the communication does work, it is responsibility of the receiver code to free anything related to the task, as
422  * usual. More details on this can be obtained on
423  * <a href="http://lists.gforge.inria.fr/pipermail/simgrid-user/2011-November/002649.html">this thread</a>
424  * in the SimGrid-user mailing list archive.
425  *
426  * \param task a #msg_task_t to send on another location.
427  * \param alias name of the mailbox to sent the task to
428  * \param cleanup a function to destroy the task if the
429  * communication fails, e.g. MSG_task_destroy
430  * (if NULL, no function will be called)
431  * \param maxrate the maximum communication rate for sending this task
432  *
433  */
434 void MSG_task_dsend_bounded(msg_task_t task, const char *alias, void_f_pvoid_t cleanup, double maxrate)
435 {
436   task->simdata->rate = maxrate;
437   MSG_task_dsend(task, alias, cleanup);
438 }
439
440 /** \ingroup msg_task_usage
441  * \brief Starts listening for receiving a task from an asynchronous communication.
442  *
443  * This is a non blocking function: use MSG_comm_wait() or MSG_comm_test() to end the communication.
444  *
445  * \param task a memory location for storing a #msg_task_t. has to be valid until the end of the communication.
446  * \param name of the mailbox to receive the task on
447  * \return the msg_comm_t communication created
448  */
449 msg_comm_t MSG_task_irecv(msg_task_t *task, const char *name)
450 {
451   return MSG_task_irecv_bounded(task, name, -1.0);
452 }
453
454 /** \ingroup msg_task_usage
455  * \brief Starts listening for receiving a task from an asynchronous communication at a given rate.
456  *
457  * \param task a memory location for storing a #msg_task_t. has to be valid until the end of the communication.
458  * \param name of the mailbox to receive the task on
459  * \param rate limit the bandwidth to the given rate
460  * \return the msg_comm_t communication created
461  */
462 msg_comm_t MSG_task_irecv_bounded(msg_task_t *task, const char *name, double rate)
463 {
464   smx_mailbox_t rdv = MSG_mailbox_get_by_alias(name);
465
466   /* FIXME: these functions are not traceable */
467   /* Sanity check */
468   xbt_assert(task, "Null pointer for the task storage");
469
470   if (*task)
471     XBT_CRITICAL("MSG_task_irecv() was asked to write in a non empty task struct.");
472
473   /* Try to receive it by calling SIMIX network layer */
474   msg_comm_t comm = xbt_new0(s_msg_comm_t, 1);
475   comm->task_sent = NULL;
476   comm->task_received = task;
477   comm->status = MSG_OK;
478   comm->s_comm = simcall_comm_irecv(MSG_process_self(), rdv, task, NULL, NULL, NULL, NULL, rate);
479
480   return comm;
481 }
482
483 /** \ingroup msg_task_usage
484  * \brief Checks whether a communication is done, and if yes, finalizes it.
485  * \param comm the communication to test
486  * \return TRUE if the communication is finished
487  * (but it may have failed, use MSG_comm_get_status() to know its status)
488  * or FALSE if the communication is not finished yet
489  * If the status is FALSE, don't forget to use MSG_process_sleep() after the test.
490  */
491 int MSG_comm_test(msg_comm_t comm)
492 {
493   xbt_ex_t e;
494   int finished = 0;
495
496   TRY {
497     finished = simcall_comm_test(comm->s_comm);
498
499     if (finished && comm->task_received != NULL) {
500       /* I am the receiver */
501       if (msg_global->debug_multiple_use && (*comm->task_received)->simdata->isused!=0)
502         xbt_ex_free(*(xbt_ex_t*)(*comm->task_received)->simdata->isused);
503       (*comm->task_received)->simdata->isused = 0;
504     }
505   }
506   CATCH(e) {
507     switch (e.category) {
508       case network_error:
509         comm->status = MSG_TRANSFER_FAILURE;
510         finished = 1;
511         break;
512       case timeout_error:
513         comm->status = MSG_TIMEOUT;
514         finished = 1;
515         break;
516       default:
517         RETHROW;
518     }
519     xbt_ex_free(e);
520   }
521
522   return finished;
523 }
524
525 /** \ingroup msg_task_usage
526  * \brief This function checks if a communication is finished.
527  * \param comms a vector of communications
528  * \return the position of the finished communication if any
529  * (but it may have failed, use MSG_comm_get_status() to know its status),
530  * or -1 if none is finished
531  */
532 int MSG_comm_testany(xbt_dynar_t comms)
533 {
534   xbt_ex_t e;
535   int finished_index = -1;
536
537   /* create the equivalent dynar with SIMIX objects */
538   xbt_dynar_t s_comms = xbt_dynar_new(sizeof(smx_synchro_t), NULL);
539   msg_comm_t comm;
540   unsigned int cursor;
541   xbt_dynar_foreach(comms, cursor, comm) {
542     xbt_dynar_push(s_comms, &comm->s_comm);
543   }
544
545   msg_error_t status = MSG_OK;
546   TRY {
547     finished_index = simcall_comm_testany(s_comms);
548   }
549   CATCH(e) {
550     switch (e.category) {
551       case network_error:
552         finished_index = e.value;
553         status = MSG_TRANSFER_FAILURE;
554         break;
555       case timeout_error:
556         finished_index = e.value;
557         status = MSG_TIMEOUT;
558         break;
559       default:
560         RETHROW;
561     }
562     xbt_ex_free(e);
563   }
564   xbt_dynar_free(&s_comms);
565
566   if (finished_index != -1) {
567     comm = xbt_dynar_get_as(comms, finished_index, msg_comm_t);
568     /* the communication is finished */
569     comm->status = status;
570
571     if (status == MSG_OK && comm->task_received != NULL) {
572       /* I am the receiver */
573       if (msg_global->debug_multiple_use && (*comm->task_received)->simdata->isused!=0)
574         xbt_ex_free(*(xbt_ex_t*)(*comm->task_received)->simdata->isused);
575       (*comm->task_received)->simdata->isused = 0;
576     }
577   }
578
579   return finished_index;
580 }
581
582 /** \ingroup msg_task_usage
583  * \brief Destroys a communication.
584  * \param comm the communication to destroy.
585  */
586 void MSG_comm_destroy(msg_comm_t comm)
587 {
588   xbt_free(comm);
589 }
590
591 /** \ingroup msg_task_usage
592  * \brief Wait for the completion of a communication.
593  *
594  * It takes two parameters.
595  * \param comm the communication to wait.
596  * \param timeout Wait until the communication terminates or the timeout occurs.
597  *                You can provide a -1 timeout to obtain an infinite timeout.
598  * \return msg_error_t
599  */
600 msg_error_t MSG_comm_wait(msg_comm_t comm, double timeout)
601 {
602   xbt_ex_t e;
603   TRY {
604     simcall_comm_wait(comm->s_comm, timeout);
605
606     if (comm->task_received != NULL) {
607       /* I am the receiver */
608       if (msg_global->debug_multiple_use && (*comm->task_received)->simdata->isused!=0)
609         xbt_ex_free(*(xbt_ex_t*)(*comm->task_received)->simdata->isused);
610       (*comm->task_received)->simdata->isused = 0;
611     }
612
613     /* FIXME: these functions are not traceable */
614   }
615   CATCH(e) {
616     switch (e.category) {
617     case network_error:
618       comm->status = MSG_TRANSFER_FAILURE;
619       break;
620     case timeout_error:
621       comm->status = MSG_TIMEOUT;
622       break;
623     default:
624       RETHROW;
625     }
626     xbt_ex_free(e);
627   }
628
629   return comm->status;
630 }
631
632 /** \ingroup msg_task_usage
633 * \brief This function is called by a sender and permit to wait for each communication
634 *
635 * \param comm a vector of communication
636 * \param nb_elem is the size of the comm vector
637 * \param timeout for each call of MSG_comm_wait
638 */
639 void MSG_comm_waitall(msg_comm_t * comm, int nb_elem, double timeout)
640 {
641   int i = 0;
642   for (i = 0; i < nb_elem; i++) {
643     MSG_comm_wait(comm[i], timeout);
644   }
645 }
646
647 /** \ingroup msg_task_usage
648  * \brief This function waits for the first communication finished in a list.
649  * \param comms a vector of communications
650  * \return the position of the first finished communication
651  * (but it may have failed, use MSG_comm_get_status() to know its status)
652  */
653 int MSG_comm_waitany(xbt_dynar_t comms)
654 {
655   xbt_ex_t e;
656   int finished_index = -1;
657
658   /* create the equivalent dynar with SIMIX objects */
659   xbt_dynar_t s_comms = xbt_dynar_new(sizeof(smx_synchro_t), NULL);
660   msg_comm_t comm;
661   unsigned int cursor;
662   xbt_dynar_foreach(comms, cursor, comm) {
663     xbt_dynar_push(s_comms, &comm->s_comm);
664   }
665
666   msg_error_t status = MSG_OK;
667   TRY {
668     finished_index = simcall_comm_waitany(s_comms);
669   }
670   CATCH(e) {
671     switch (e.category) {
672       case network_error:
673         finished_index = e.value;
674         status = MSG_TRANSFER_FAILURE;
675         break;
676       case timeout_error:
677         finished_index = e.value;
678         status = MSG_TIMEOUT;
679         break;
680       default:
681         RETHROW;
682     }
683     xbt_ex_free(e);
684   }
685
686   xbt_assert(finished_index != -1, "WaitAny returned -1");
687   xbt_dynar_free(&s_comms);
688
689   comm = xbt_dynar_get_as(comms, finished_index, msg_comm_t);
690   /* the communication is finished */
691   comm->status = status;
692
693   if (comm->task_received != NULL) {
694     /* I am the receiver */
695     if (msg_global->debug_multiple_use && (*comm->task_received)->simdata->isused!=0)
696       xbt_ex_free(*(xbt_ex_t*)(*comm->task_received)->simdata->isused);
697     (*comm->task_received)->simdata->isused = 0;
698   }
699
700   return finished_index;
701 }
702
703 /**
704  * \ingroup msg_task_usage
705  * \brief Returns the error (if any) that occured during a finished communication.
706  * \param comm a finished communication
707  * \return the status of the communication, or #MSG_OK if no error occured
708  * during the communication
709  */
710 msg_error_t MSG_comm_get_status(msg_comm_t comm) {
711
712   return comm->status;
713 }
714
715 /** \ingroup msg_task_usage
716  * \brief Get a task (#msg_task_t) from a communication
717  *
718  * \param comm the communication where to get the task
719  * \return the task from the communication
720  */
721 msg_task_t MSG_comm_get_task(msg_comm_t comm)
722 {
723   xbt_assert(comm, "Invalid parameter");
724
725   return comm->task_received ? *comm->task_received : comm->task_sent;
726 }
727
728 /**
729  * \brief This function is called by SIMIX in kernel mode to copy the data of a comm.
730  * \param comm the comm
731  * \param buff the data copied
732  * \param buff_size size of the buffer
733  */
734 void MSG_comm_copy_data_from_SIMIX(smx_synchro_t comm, void* buff, size_t buff_size) {
735   // copy the task
736   SIMIX_comm_copy_pointer_callback(comm, buff, buff_size);
737
738   // notify the user callback if any
739   if (msg_global->task_copy_callback) {
740     msg_task_t task = (msg_task_t) buff;
741     msg_global->task_copy_callback(task, simcall_comm_get_src_proc(comm), simcall_comm_get_dst_proc(comm));
742   }
743 }
744
745 /** \ingroup msg_task_usage
746  * \brief Sends a task to a mailbox
747  *
748  * This is a blocking function, the execution flow will be blocked until the task is sent (and received on the other
749  * side if #MSG_task_receive is used).
750  * See #MSG_task_isend for sending tasks asynchronously.
751  *
752  * \param task the task to be sent
753  * \param alias the mailbox name to where the task is sent
754  *
755  * \return Returns #MSG_OK if the task was successfully sent,
756  * #MSG_HOST_FAILURE, or #MSG_TRANSFER_FAILURE otherwise.
757  */
758 msg_error_t MSG_task_send(msg_task_t task, const char *alias)
759 {
760   XBT_DEBUG("MSG_task_send: Trying to send a message on mailbox '%s'", alias);
761   return MSG_task_send_with_timeout(task, alias, -1);
762 }
763
764 /** \ingroup msg_task_usage
765  * \brief Sends a task to a mailbox with a maximum rate
766  *
767  * This is a blocking function, the execution flow will be blocked until the task is sent. The maxrate parameter allows
768  * the application to limit the bandwidth utilization of network links when sending the task.
769  *
770  * \param task the task to be sent
771  * \param alias the mailbox name to where the task is sent
772  * \param maxrate the maximum communication rate for sending this task
773  *
774  * \return Returns #MSG_OK if the task was successfully sent,
775  * #MSG_HOST_FAILURE, or #MSG_TRANSFER_FAILURE otherwise.
776  */
777 msg_error_t MSG_task_send_bounded(msg_task_t task, const char *alias, double maxrate)
778 {
779   task->simdata->rate = maxrate;
780   return MSG_task_send(task, alias);
781 }
782
783 /** \ingroup msg_task_usage
784  * \brief Sends a task to a mailbox with a timeout
785  *
786  * This is a blocking function, the execution flow will be blocked until the task is sent or the timeout is achieved.
787  *
788  * \param task the task to be sent
789  * \param alias the mailbox name to where the task is sent
790  * \param timeout is the maximum wait time for completion (if -1, this call is the same as #MSG_task_send)
791  *
792  * \return Returns #MSG_OK if the task was successfully sent,
793  * #MSG_HOST_FAILURE, or #MSG_TRANSFER_FAILURE, or #MSG_TIMEOUT otherwise.
794  */
795 msg_error_t MSG_task_send_with_timeout(msg_task_t task, const char *alias, double timeout)
796 {
797   return MSG_mailbox_put_with_timeout(MSG_mailbox_get_by_alias(alias), task, timeout);
798 }
799
800 /** \ingroup msg_task_usage
801  * \brief Sends a task to a mailbox with a timeout and with a maximum rate
802  *
803  * This is a blocking function, the execution flow will be blocked until the task is sent or the timeout is achieved.
804  *
805  * \param task the task to be sent
806  * \param alias the mailbox name to where the task is sent
807  * \param timeout is the maximum wait time for completion (if -1, this call is the same as #MSG_task_send)
808  * \param maxrate the maximum communication rate for sending this task
809  *
810  * \return Returns #MSG_OK if the task was successfully sent,
811  * #MSG_HOST_FAILURE, or #MSG_TRANSFER_FAILURE, or #MSG_TIMEOUT otherwise.
812  */
813 msg_error_t MSG_task_send_with_timeout_bounded(msg_task_t task, const char *alias, double timeout, double maxrate)
814 {
815   task->simdata->rate = maxrate;
816   return MSG_mailbox_put_with_timeout(MSG_mailbox_get_by_alias(alias), task, timeout);
817 }
818
819 /** \ingroup msg_task_usage
820  * \brief Check if there is a communication going on in a mailbox.
821  *
822  * \param alias the name of the mailbox to be considered
823  *
824  * \return Returns 1 if there is a communication, 0 otherwise
825  */
826 int MSG_task_listen(const char *alias)
827 {
828   smx_mailbox_t mbox = MSG_mailbox_get_by_alias(alias);
829   return !MSG_mailbox_is_empty(mbox) || (mbox->permanent_receiver && !mbox->done_comm_queue->empty());
830 }
831
832 /** \ingroup msg_task_usage
833  * \brief Look if there is a communication on a mailbox and return the PID of the sender process.
834  *
835  * \param alias the name of the mailbox to be considered
836  *
837  * \return Returns the PID of sender process,
838  * -1 if there is no communication in the mailbox.
839  */
840 int MSG_task_listen_from(const char *alias)
841 {
842   msg_task_t task;
843
844   if (NULL == (task = MSG_mailbox_get_head(MSG_mailbox_get_by_alias(alias))))
845     return -1;
846
847   return MSG_process_get_PID(task->simdata->sender);
848 }
849
850 /** \ingroup msg_task_usage
851  * \brief Sets the tracing category of a task.
852  *
853  * This function should be called after the creation of a MSG task, to define the category of that task. The
854  * first parameter task must contain a task that was  created with the function #MSG_task_create. The second
855  * parameter category must contain a category that was previously declared with the function #TRACE_category
856  * (or with #TRACE_category_with_color).
857  *
858  * See \ref tracing for details on how to trace the (categorized) resource utilization.
859  *
860  * \param task the task that is going to be categorized
861  * \param category the name of the category to be associated to the task
862  *
863  * \see MSG_task_get_category, TRACE_category, TRACE_category_with_color
864  */
865 void MSG_task_set_category (msg_task_t task, const char *category)
866 {
867   TRACE_msg_set_task_category (task, category);
868 }
869
870 /** \ingroup msg_task_usage
871  *
872  * \brief Gets the current tracing category of a task.
873  *
874  * \param task the task to be considered
875  *
876  * \see MSG_task_set_category
877  *
878  * \return Returns the name of the tracing category of the given task, NULL otherwise
879  */
880 const char *MSG_task_get_category (msg_task_t task)
881 {
882   return task->category;
883 }
884
885 /**
886  * \brief Returns the value of a given AS or router property
887  *
888  * \param asr the name of a router or AS
889  * \param name a property name
890  * \return value of a property (or NULL if property not set)
891  */
892 const char *MSG_as_router_get_property_value(const char* asr, const char *name)
893 {
894   return (char*) xbt_dict_get_or_null(MSG_as_router_get_properties(asr), name);
895 }
896
897 /**
898  * \brief Returns a xbt_dict_t consisting of the list of properties assigned to
899  * a the AS or router
900  *
901  * \param asr the name of a router or AS
902  * \return a dict containing the properties
903  */
904 xbt_dict_t MSG_as_router_get_properties(const char* asr)
905 {
906   return (simcall_asr_get_properties(asr));
907 }
908
909 /**
910  * \brief Change the value of a given AS or router
911  *
912  * \param asr the name of a router or AS
913  * \param name a property name
914  * \param value what to change the property to
915  * \param free_ctn the freeing function to use to kill the value on need
916  */
917 void MSG_as_router_set_property_value(const char* asr, const char *name, char *value,void_f_pvoid_t free_ctn) {
918   xbt_dict_set(MSG_as_router_get_properties(asr), name, value,free_ctn);
919 }