Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
forgot that one
[simgrid.git] / src / msg / msg_gos.cpp
1 /* Copyright (c) 2004-2019. 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 "simgrid/Exception.hpp"
7 #include <cmath>
8
9 #include "simgrid/s4u/Comm.hpp"
10 #include "simgrid/s4u/Mailbox.hpp"
11 #include "src/instr/instr_private.hpp"
12 #include "src/kernel/activity/ExecImpl.hpp"
13 #include "src/msg/msg_private.hpp"
14 #include "src/simix/smx_private.hpp" /* MSG_task_listen looks inside the rdv directly. Not clean. */
15
16 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(msg_gos, msg, "Logging specific to MSG (gos)");
17
18 /**
19  * @brief Executes a task and waits for its termination.
20  *
21  * This function is used for describing the behavior of a process. It takes only one parameter.
22  * @param task a #msg_task_t to execute on the location on which the process is running.
23  * @return #MSG_OK if the task was successfully completed, #MSG_TASK_CANCELED or #MSG_HOST_FAILURE otherwise
24  */
25 msg_error_t MSG_task_execute(msg_task_t task)
26 {
27   return MSG_parallel_task_execute(task);
28 }
29
30 /**
31  * @brief Executes a parallel task and waits for its termination.
32  *
33  * @param task a #msg_task_t to execute on the location on which the process is running.
34  *
35  * @return #MSG_OK if the task was successfully completed, #MSG_TASK_CANCELED
36  * or #MSG_HOST_FAILURE otherwise
37  */
38 msg_error_t MSG_parallel_task_execute(msg_task_t task)
39 {
40   return MSG_parallel_task_execute_with_timeout(task, -1);
41 }
42
43 msg_error_t MSG_parallel_task_execute_with_timeout(msg_task_t task, double timeout)
44 {
45   simdata_task_t simdata = task->simdata;
46   e_smx_state_t comp_state;
47   msg_error_t status = MSG_OK;
48
49   xbt_assert((not simdata->compute) && not task->simdata->is_used,
50              "This task is executed somewhere else. Go fix your code!");
51
52   XBT_DEBUG("Computing on %s", MSG_process_get_name(MSG_process_self()));
53
54   if (simdata->flops_amount <= 0.0 && not simdata->host_nb) {
55     return MSG_OK;
56   }
57
58   if (TRACE_actor_is_enabled())
59     simgrid::instr::Container::by_name(instr_pid(MSG_process_self()))->get_state("ACTOR_STATE")->push_event("execute");
60
61   try {
62     simdata->set_used();
63
64     if (simdata->host_nb > 0) {
65       simdata->compute =
66           boost::static_pointer_cast<simgrid::kernel::activity::ExecImpl>(simcall_execution_parallel_start(
67               task->name ?: "", simdata->host_nb, simdata->host_list, simdata->flops_parallel_amount,
68               simdata->bytes_parallel_amount, -1.0, timeout));
69       XBT_DEBUG("Parallel execution action created: %p", simdata->compute.get());
70       if (task->category != nullptr)
71         simgrid::simix::simcall([task] { task->simdata->compute->set_category(task->category); });
72     } else {
73       sg_host_t host   = MSG_process_get_host(MSG_process_self());
74       simdata->compute = simgrid::simix::simcall([task, host] {
75         return simgrid::kernel::activity::ExecImplPtr(
76             new simgrid::kernel::activity::ExecImpl(task->name ?: "", task->category ?: "",
77                                                     /*timeout_detector*/ nullptr, host));
78       });
79       /* checking for infinite values */
80       xbt_assert(std::isfinite(simdata->flops_amount), "flops_amount is not finite!");
81       xbt_assert(std::isfinite(simdata->priority), "priority is not finite!");
82
83       simdata->compute->start(simdata->flops_amount, simdata->priority, simdata->bound);
84     }
85
86     comp_state = simcall_execution_wait(simdata->compute);
87
88     simdata->set_not_used();
89
90     XBT_DEBUG("Execution task '%s' finished in state %d", task->name, (int)comp_state);
91   } catch (simgrid::HostFailureException& e) {
92     status = MSG_HOST_FAILURE;
93   } catch (simgrid::TimeoutError& e) {
94     status = MSG_TIMEOUT;
95   } catch (simgrid::CancelException& e) {
96     status = MSG_TASK_CANCELED;
97   }
98
99   /* action ended, set comm and compute = nullptr, the actions is already destroyed in the main function */
100   simdata->flops_amount = 0.0;
101   simdata->comm = nullptr;
102   simdata->compute = nullptr;
103
104   if (TRACE_actor_is_enabled())
105     simgrid::instr::Container::by_name(instr_pid(MSG_process_self()))->get_state("ACTOR_STATE")->pop_event();
106
107   return status;
108 }
109
110 /**
111  * @brief Receives a task from a mailbox.
112  *
113  * This is a blocking function, the execution flow will be blocked until the task is received. See #MSG_task_irecv
114  * for receiving tasks asynchronously.
115  *
116  * @param task a memory location for storing a #msg_task_t.
117  * @param alias name of the mailbox to receive the task from
118  *
119  * @return Returns
120  * #MSG_OK if the task was successfully received,
121  * #MSG_HOST_FAILURE, or #MSG_TRANSFER_FAILURE otherwise.
122  */
123 msg_error_t MSG_task_receive(msg_task_t * task, const char *alias)
124 {
125   return MSG_task_receive_with_timeout(task, alias, -1);
126 }
127
128 /**
129  * @brief Receives a task from a mailbox at a given rate.
130  *
131  * @param task a memory location for storing a #msg_task_t.
132  * @param alias name of the mailbox to receive the task from
133  * @param rate limit the reception to rate bandwidth (byte/sec)
134  *
135  * The rate parameter can be used to receive a task with a limited
136  * bandwidth (smaller than the physical available value). Use
137  * MSG_task_receive() if you don't limit the rate (or pass -1 as a
138  * rate value do disable this feature).
139  *
140  * @return Returns
141  * #MSG_OK if the task was successfully received,
142  * #MSG_HOST_FAILURE, or #MSG_TRANSFER_FAILURE otherwise.
143  */
144 msg_error_t MSG_task_receive_bounded(msg_task_t * task, const char *alias, double rate)
145 {
146   return MSG_task_receive_with_timeout_bounded(task, alias, -1, rate);
147 }
148
149 /**
150  * @brief Receives a task from a mailbox with a given timeout.
151  *
152  * This is a blocking function with a timeout, the execution flow will be blocked until the task is received or the
153  * timeout is achieved. See #MSG_task_irecv for receiving tasks asynchronously.  You can provide a -1 timeout
154  * to obtain an infinite timeout.
155  *
156  * @param task a memory location for storing a #msg_task_t.
157  * @param alias name of the mailbox to receive the task from
158  * @param timeout is the maximum wait time for completion (if -1, this call is the same as #MSG_task_receive)
159  *
160  * @return Returns
161  * #MSG_OK if the task was successfully received,
162  * #MSG_HOST_FAILURE, or #MSG_TRANSFER_FAILURE, or #MSG_TIMEOUT otherwise.
163  */
164 msg_error_t MSG_task_receive_with_timeout(msg_task_t * task, const char *alias, double timeout)
165 {
166   return MSG_task_receive_ext(task, alias, timeout, nullptr);
167 }
168
169 /**
170  * @brief Receives a task from a mailbox with a given timeout and at a given rate.
171  *
172  * @param task a memory location for storing a #msg_task_t.
173  * @param alias name of the mailbox to receive the task from
174  * @param timeout is the maximum wait time for completion (if -1, this call is the same as #MSG_task_receive)
175  * @param rate limit the reception to rate bandwidth (byte/sec)
176  *
177  * The rate parameter can be used to send a task with a limited
178  * bandwidth (smaller than the physical available value). Use
179  * MSG_task_receive() if you don't limit the rate (or pass -1 as a
180  * rate value do disable this feature).
181  *
182  * @return Returns
183  * #MSG_OK if the task was successfully received,
184  * #MSG_HOST_FAILURE, or #MSG_TRANSFER_FAILURE, or #MSG_TIMEOUT otherwise.
185  */
186 msg_error_t MSG_task_receive_with_timeout_bounded(msg_task_t * task, const char *alias, double timeout,double rate)
187 {
188   return MSG_task_receive_ext_bounded(task, alias, timeout, nullptr, rate);
189 }
190
191 /**
192  * @brief Receives a task from a mailbox from a specific host with a given timeout.
193  *
194  * This is a blocking function with a timeout, the execution flow will be blocked until the task is received or the
195  * timeout is achieved. See #MSG_task_irecv for receiving tasks asynchronously. You can provide a -1 timeout
196  * to obtain an infinite timeout.
197  *
198  * @param task a memory location for storing a #msg_task_t.
199  * @param alias name of the mailbox to receive the task from
200  * @param timeout is the maximum wait time for completion (provide -1 for no timeout)
201  * @param host a #msg_host_t host from where the task was sent
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_ext(msg_task_t * task, const char *alias, double timeout, msg_host_t host)
208 {
209   XBT_DEBUG("MSG_task_receive_ext: Trying to receive a message on mailbox '%s'", alias);
210   return MSG_task_receive_ext_bounded(task, alias, timeout, host, -1.0);
211 }
212
213 /**
214  * @brief Receives a task from a mailbox from a specific host with a given timeout  and at a given rate.
215  *
216  * @param task a memory location for storing a #msg_task_t.
217  * @param alias name of the mailbox to receive the task from
218  * @param timeout is the maximum wait time for completion (provide -1 for no timeout)
219  * @param host a #msg_host_t host from where the task was sent
220  * @param rate limit the reception to rate bandwidth (byte/sec)
221  *
222  * The rate parameter can be used to receive a task with a limited
223  * bandwidth (smaller than the physical available value). Use
224  * MSG_task_receive_ext() if you don't limit the rate (or pass -1 as a
225  * rate value do disable this feature).
226  *
227  * @return Returns
228  * #MSG_OK if the task was successfully received,
229  * #MSG_HOST_FAILURE, or #MSG_TRANSFER_FAILURE, or #MSG_TIMEOUT otherwise.
230  */
231 msg_error_t MSG_task_receive_ext_bounded(msg_task_t * task, const char *alias, double timeout, msg_host_t host,
232                                          double rate)
233 {
234   XBT_DEBUG("MSG_task_receive_ext: Trying to receive a message on mailbox '%s'", alias);
235   msg_error_t ret = MSG_OK;
236   /* We no longer support getting a task from a specific host */
237   if (host)
238     THROW_UNIMPLEMENTED;
239
240   /* Sanity check */
241   xbt_assert(task, "Null pointer for the task storage");
242
243   if (*task)
244     XBT_WARN("Asked to write the received task in a non empty struct -- proceeding.");
245
246   /* Try to receive it by calling SIMIX network layer */
247   try {
248     void* payload;
249     simgrid::s4u::Mailbox::by_name(alias)
250         ->get_init()
251         ->set_dst_data(&payload, sizeof(msg_task_t*))
252         ->set_rate(rate)
253         ->wait_for(timeout);
254     *task = static_cast<msg_task_t>(payload);
255     XBT_DEBUG("Got task %s from %s", (*task)->name, alias);
256     (*task)->simdata->set_not_used();
257   } catch (simgrid::HostFailureException& e) {
258     ret = MSG_HOST_FAILURE;
259   } catch (simgrid::TimeoutError& e) {
260     ret = MSG_TIMEOUT;
261   } catch (simgrid::CancelException& e) {
262     ret = MSG_TASK_CANCELED;
263   } catch (xbt_ex& e) {
264     if (e.category == network_error)
265       ret = MSG_TRANSFER_FAILURE;
266     else
267       throw;
268   }
269
270   if (TRACE_actor_is_enabled() && ret != MSG_HOST_FAILURE && ret != MSG_TRANSFER_FAILURE && ret != MSG_TIMEOUT) {
271     container_t process_container = simgrid::instr::Container::by_name(instr_pid(MSG_process_self()));
272
273     std::string key = std::string("p") + std::to_string((*task)->counter);
274     simgrid::instr::Container::get_root()->get_link("ACTOR_TASK_LINK")->end_event(process_container, "SR", key);
275   }
276   return ret;
277 }
278
279 /* Internal function used to factorize code between MSG_task_isend(), MSG_task_isend_bounded(), and MSG_task_dsend(). */
280 static inline msg_comm_t MSG_task_isend_internal(msg_task_t task, const char* alias, void_f_pvoid_t cleanup,
281                                                  bool detached)
282 {
283   simdata_task_t t_simdata = nullptr;
284   msg_process_t myself = MSG_process_self();
285   simgrid::s4u::MailboxPtr mailbox = simgrid::s4u::Mailbox::by_name(alias);
286   TRACE_msg_task_put_start(task);
287
288   /* Prepare the task to send */
289   t_simdata = task->simdata;
290   t_simdata->sender = myself;
291   t_simdata->source = MSG_host_self();
292   t_simdata->set_used();
293   t_simdata->comm = nullptr;
294   msg_global->sent_msg++;
295
296   simgrid::s4u::CommPtr comm = mailbox->put_init(task, t_simdata->bytes_amount)->set_rate(t_simdata->rate);
297   t_simdata->comm            = comm;
298   if (detached)
299     comm->detach(cleanup);
300   else
301     comm->start();
302
303   msg_comm_t msg_comm = nullptr;
304   if (not detached) {
305     msg_comm = new simgrid::msg::Comm(task, nullptr, comm);
306   }
307
308   if (TRACE_is_enabled() && task->category != nullptr)
309     simgrid::simix::simcall([comm, task] { comm->get_impl()->set_category(task->category); });
310
311   return msg_comm;
312 }
313
314 /**
315  * @brief Sends a task on a mailbox.
316  *
317  * This is a non blocking function: use MSG_comm_wait() or MSG_comm_test() to end the communication.
318  *
319  * @param task a #msg_task_t to send on another location.
320  * @param alias name of the mailbox to sent the task to
321  * @return the msg_comm_t communication created
322  */
323 msg_comm_t MSG_task_isend(msg_task_t task, const char *alias)
324 {
325   return MSG_task_isend_internal(task, alias, nullptr, false);
326 }
327
328 /**
329  * @brief Sends a task on a mailbox with a maximum rate
330  *
331  * This is a non blocking function: use MSG_comm_wait() or MSG_comm_test() to end the communication. The maxrate
332  * parameter allows the application to limit the bandwidth utilization of network links when sending the task.
333  *
334  * @param task a #msg_task_t to send on another location.
335  * @param alias name of the mailbox to sent the task to
336  * @param maxrate the maximum communication rate for sending this task (byte/sec).
337  * @return the msg_comm_t communication created
338  */
339 msg_comm_t MSG_task_isend_bounded(msg_task_t task, const char *alias, double maxrate)
340 {
341   task->simdata->rate = maxrate;
342   return MSG_task_isend_internal(task, alias, nullptr, false);
343 }
344
345 /**
346  * @brief Sends a task on a mailbox.
347  *
348  * This is a non blocking detached send function.
349  * Think of it as a best effort send. Keep in mind that the third parameter is only called if the communication fails.
350  * If the communication does work, it is responsibility of the receiver code to free anything related to the task, as
351  * usual. More details on this can be obtained on
352  * <a href="http://lists.gforge.inria.fr/pipermail/simgrid-user/2011-November/002649.html">this thread</a>
353  * in the SimGrid-user mailing list archive.
354  *
355  * @param task a #msg_task_t to send on another location.
356  * @param alias name of the mailbox to sent the task to
357  * @param cleanup a function to destroy the task if the communication fails, e.g. MSG_task_destroy
358  * (if nullptr, no function will be called)
359  */
360 void MSG_task_dsend(msg_task_t task, const char *alias, void_f_pvoid_t cleanup)
361 {
362   msg_comm_t XBT_ATTRIB_UNUSED comm = MSG_task_isend_internal(task, alias, cleanup, true);
363   xbt_assert(comm == nullptr);
364 }
365
366 /**
367  * @brief Sends a task on a mailbox with a maximal rate.
368  *
369  * This is a non blocking detached send function.
370  * Think of it as a best effort send. Keep in mind that the third parameter is only called if the communication fails.
371  * If the communication does work, it is responsibility of the receiver code to free anything related to the task, as
372  * usual. More details on this can be obtained on
373  * <a href="http://lists.gforge.inria.fr/pipermail/simgrid-user/2011-November/002649.html">this thread</a>
374  * in the SimGrid-user mailing list archive.
375  *
376  * The rate parameter can be used to send a task with a limited
377  * bandwidth (smaller than the physical available value). Use
378  * MSG_task_dsend() if you don't limit the rate (or pass -1 as a rate
379  * value do disable this feature).
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 cleanup a function to destroy the task if the
384  * communication fails, e.g. MSG_task_destroy
385  * (if nullptr, no function will be called)
386  * @param maxrate the maximum communication rate for sending this task (byte/sec)
387  *
388  */
389 void MSG_task_dsend_bounded(msg_task_t task, const char *alias, void_f_pvoid_t cleanup, double maxrate)
390 {
391   task->simdata->rate = maxrate;
392   MSG_task_dsend(task, alias, cleanup);
393 }
394
395 /**
396  * @brief Starts listening for receiving a task from an asynchronous communication.
397  *
398  * This is a non blocking function: use MSG_comm_wait() or MSG_comm_test() to end the communication.
399  *
400  * @param task a memory location for storing a #msg_task_t. has to be valid until the end of the communication.
401  * @param name of the mailbox to receive the task on
402  * @return the msg_comm_t communication created
403  */
404 msg_comm_t MSG_task_irecv(msg_task_t *task, const char *name)
405 {
406   return MSG_task_irecv_bounded(task, name, -1.0);
407 }
408
409 /**
410  * @brief Starts listening for receiving a task from an asynchronous communication at a given rate.
411  *
412  * The rate parameter can be used to receive a task with a limited
413  * bandwidth (smaller than the physical available value). Use
414  * MSG_task_irecv() if you don't limit the rate (or pass -1 as a rate
415  * value do disable this feature).
416  *
417  * @param task a memory location for storing a #msg_task_t. has to be valid until the end of the communication.
418  * @param name of the mailbox to receive the task on
419  * @param rate limit the bandwidth to the given rate (byte/sec)
420  * @return the msg_comm_t communication created
421  */
422 msg_comm_t MSG_task_irecv_bounded(msg_task_t *task, const char *name, double rate)
423 {
424   simgrid::s4u::MailboxPtr mbox = simgrid::s4u::Mailbox::by_name(name);
425
426   /* FIXME: these functions are not traceable */
427   /* Sanity check */
428   xbt_assert(task, "Null pointer for the task storage");
429
430   if (*task)
431     XBT_CRITICAL("MSG_task_irecv() was asked to write in a non empty task struct.");
432
433   /* Try to receive it by calling SIMIX network layer */
434   msg_comm_t comm = new simgrid::msg::Comm(
435       nullptr, task, mbox->get_init()->set_dst_data((void**)task, sizeof(msg_task_t*))->set_rate(rate)->start());
436
437   return comm;
438 }
439
440 /**
441  * @brief Checks whether a communication is done, and if yes, finalizes it.
442  * @param comm the communication to test
443  * @return 'true' if the communication is finished
444  * (but it may have failed, use MSG_comm_get_status() to know its status)
445  * or 'false' if the communication is not finished yet
446  * If the status is 'false', don't forget to use MSG_process_sleep() after the test.
447  */
448 int MSG_comm_test(msg_comm_t comm)
449 {
450   bool finished = false;
451
452   try {
453     finished = comm->s_comm->test();
454     if (finished && comm->task_received != nullptr) {
455       /* I am the receiver */
456       (*comm->task_received)->simdata->set_not_used();
457     }
458   } catch (simgrid::TimeoutError& e) {
459     comm->status = MSG_TIMEOUT;
460     finished     = true;
461   } catch (simgrid::CancelException& e) {
462     comm->status = MSG_TASK_CANCELED;
463     finished     = true;
464   }
465   catch (xbt_ex& e) {
466     if (e.category == network_error) {
467       comm->status = MSG_TRANSFER_FAILURE;
468       finished     = true;
469     } else {
470       throw;
471     }
472   }
473
474   return finished;
475 }
476
477 /**
478  * @brief This function checks if a communication is finished.
479  * @param comms a vector of communications
480  * @return the position of the finished communication if any
481  * (but it may have failed, use MSG_comm_get_status() to know its status),
482  * or -1 if none is finished
483  */
484 int MSG_comm_testany(xbt_dynar_t comms)
485 {
486   int finished_index = -1;
487
488   /* Create the equivalent array with SIMIX objects: */
489   std::vector<simgrid::kernel::activity::CommImpl*> s_comms;
490   s_comms.reserve(xbt_dynar_length(comms));
491   msg_comm_t comm;
492   unsigned int cursor;
493   xbt_dynar_foreach(comms, cursor, comm) {
494     s_comms.push_back(static_cast<simgrid::kernel::activity::CommImpl*>(comm->s_comm->get_impl().get()));
495   }
496
497   msg_error_t status = MSG_OK;
498   try {
499     finished_index = simcall_comm_testany(s_comms.data(), s_comms.size());
500   } catch (simgrid::TimeoutError& e) {
501     finished_index = e.value;
502     status         = MSG_TIMEOUT;
503   } catch (simgrid::CancelException& e) {
504     finished_index = e.value;
505     status         = MSG_TASK_CANCELED;
506   }
507   catch (xbt_ex& e) {
508     if (e.category != network_error)
509       throw;
510     finished_index = e.value;
511     status         = MSG_TRANSFER_FAILURE;
512   }
513
514   if (finished_index != -1) {
515     comm = xbt_dynar_get_as(comms, finished_index, msg_comm_t);
516     /* the communication is finished */
517     comm->status = status;
518
519     if (status == MSG_OK && comm->task_received != nullptr) {
520       /* I am the receiver */
521       (*comm->task_received)->simdata->set_not_used();
522     }
523   }
524
525   return finished_index;
526 }
527
528 /** @brief Destroys the provided communication. */
529 void MSG_comm_destroy(msg_comm_t comm)
530 {
531   delete comm;
532 }
533
534 /** @brief Wait for the completion of a communication.
535  *
536  * It takes two parameters.
537  * @param comm the communication to wait.
538  * @param timeout Wait until the communication terminates or the timeout occurs.
539  *                You can provide a -1 timeout to obtain an infinite timeout.
540  * @return msg_error_t
541  */
542 msg_error_t MSG_comm_wait(msg_comm_t comm, double timeout)
543 {
544   try {
545     comm->s_comm->wait_for(timeout);
546
547     if (comm->task_received != nullptr) {
548       /* I am the receiver */
549       (*comm->task_received)->simdata->set_not_used();
550     }
551
552     /* FIXME: these functions are not traceable */
553   } catch (simgrid::TimeoutError& e) {
554     comm->status = MSG_TIMEOUT;
555   } catch (simgrid::CancelException& e) {
556     comm->status = MSG_TASK_CANCELED;
557   }
558   catch (xbt_ex& e) {
559     if (e.category == network_error)
560       comm->status = MSG_TRANSFER_FAILURE;
561     else
562       throw;
563   }
564
565   return comm->status;
566 }
567
568 /** @brief This function is called by a sender and permit to wait for each communication
569  *
570  * @param comm a vector of communication
571  * @param nb_elem is the size of the comm vector
572  * @param timeout for each call of MSG_comm_wait
573  */
574 void MSG_comm_waitall(msg_comm_t * comm, int nb_elem, double timeout)
575 {
576   for (int i = 0; i < nb_elem; i++)
577     MSG_comm_wait(comm[i], timeout);
578 }
579
580 /** @brief This function waits for the first communication finished in a list.
581  * @param comms a vector of communications
582  * @return the position of the first finished communication
583  * (but it may have failed, use MSG_comm_get_status() to know its status)
584  */
585 int MSG_comm_waitany(xbt_dynar_t comms)
586 {
587   int finished_index = -1;
588
589   /* Create the equivalent array with SIMIX objects: */
590   std::vector<simgrid::kernel::activity::CommImpl*> s_comms;
591   s_comms.reserve(xbt_dynar_length(comms));
592   msg_comm_t comm;
593   unsigned int cursor;
594   xbt_dynar_foreach(comms, cursor, comm) {
595     s_comms.push_back(static_cast<simgrid::kernel::activity::CommImpl*>(comm->s_comm->get_impl().get()));
596   }
597
598   msg_error_t status = MSG_OK;
599   try {
600     finished_index = simcall_comm_waitany(s_comms.data(), s_comms.size(), -1);
601   } catch (simgrid::TimeoutError& e) {
602     finished_index = e.value;
603     status         = MSG_TIMEOUT;
604   } catch (simgrid::CancelException& e) {
605     finished_index = e.value;
606     status         = MSG_TASK_CANCELED;
607   }
608   catch(xbt_ex& e) {
609     if (e.category == network_error) {
610       finished_index = e.value;
611       status         = MSG_TRANSFER_FAILURE;
612     } else {
613       throw;
614     }
615   }
616
617   xbt_assert(finished_index != -1, "WaitAny returned -1");
618
619   comm = xbt_dynar_get_as(comms, finished_index, msg_comm_t);
620   /* the communication is finished */
621   comm->status = status;
622
623   if (comm->task_received != nullptr) {
624     /* I am the receiver */
625     (*comm->task_received)->simdata->set_not_used();
626   }
627
628   return finished_index;
629 }
630
631 /**
632  * @brief Returns the error (if any) that occurred during a finished communication.
633  * @param comm a finished communication
634  * @return the status of the communication, or #MSG_OK if no error occurred
635  * during the communication
636  */
637 msg_error_t MSG_comm_get_status(msg_comm_t comm) {
638
639   return comm->status;
640 }
641
642 /** @brief Get a task (#msg_task_t) from a communication
643  *
644  * @param comm the communication where to get the task
645  * @return the task from the communication
646  */
647 msg_task_t MSG_comm_get_task(msg_comm_t comm)
648 {
649   xbt_assert(comm, "Invalid parameter");
650
651   return comm->task_received ? *comm->task_received : comm->task_sent;
652 }
653
654 /**
655  * @brief This function is called by SIMIX in kernel mode to copy the data of a comm.
656  * @param comm the comm
657  * @param buff the data copied
658  * @param buff_size size of the buffer
659  */
660 void MSG_comm_copy_data_from_SIMIX(simgrid::kernel::activity::CommImpl* comm, void* buff, size_t buff_size)
661 {
662   SIMIX_comm_copy_pointer_callback(comm, buff, buff_size);
663
664   // notify the user callback if any
665   if (msg_global->task_copy_callback) {
666     msg_task_t task = static_cast<msg_task_t>(buff);
667     msg_global->task_copy_callback(task, comm->src_actor_->ciface(), comm->dst_actor_->ciface());
668   }
669 }
670
671 /**
672  * @brief Sends a task to a mailbox
673  *
674  * This is a blocking function, the execution flow will be blocked until the task is sent (and received on the other
675  * side if #MSG_task_receive is used).
676  * See #MSG_task_isend for sending tasks asynchronously.
677  *
678  * @param task the task to be sent
679  * @param alias the mailbox name to where the task is sent
680  *
681  * @return Returns #MSG_OK if the task was successfully sent,
682  * #MSG_HOST_FAILURE, or #MSG_TRANSFER_FAILURE otherwise.
683  */
684 msg_error_t MSG_task_send(msg_task_t task, const char *alias)
685 {
686   XBT_DEBUG("MSG_task_send: Trying to send a message on mailbox '%s'", alias);
687   return MSG_task_send_with_timeout(task, alias, -1);
688 }
689
690 /**
691  * @brief Sends a task to a mailbox with a maximum rate
692  *
693  * This is a blocking function, the execution flow will be blocked until the task is sent. The maxrate parameter allows
694  * the application to limit the bandwidth utilization of network links when sending the task.
695  *
696  * The maxrate parameter can be used to send a task with a limited
697  * bandwidth (smaller than the physical available value). Use
698  * MSG_task_send() if you don't limit the rate (or pass -1 as a rate
699  * value do disable this feature).
700  *
701  * @param task the task to be sent
702  * @param alias the mailbox name to where the task is sent
703  * @param maxrate the maximum communication rate for sending this task (byte/sec)
704  *
705  * @return Returns #MSG_OK if the task was successfully sent,
706  * #MSG_HOST_FAILURE, or #MSG_TRANSFER_FAILURE otherwise.
707  */
708 msg_error_t MSG_task_send_bounded(msg_task_t task, const char *alias, double maxrate)
709 {
710   task->simdata->rate = maxrate;
711   return MSG_task_send(task, alias);
712 }
713
714 /**
715  * @brief Sends a task to a mailbox with a timeout
716  *
717  * This is a blocking function, the execution flow will be blocked until the task is sent or the timeout is achieved.
718  *
719  * @param task the task to be sent
720  * @param alias the mailbox name to where the task is sent
721  * @param timeout is the maximum wait time for completion (if -1, this call is the same as #MSG_task_send)
722  *
723  * @return Returns #MSG_OK if the task was successfully sent,
724  * #MSG_HOST_FAILURE, or #MSG_TRANSFER_FAILURE, or #MSG_TIMEOUT otherwise.
725  */
726 msg_error_t MSG_task_send_with_timeout(msg_task_t task, const char *alias, double timeout)
727 {
728   msg_error_t ret = MSG_OK;
729
730   TRACE_msg_task_put_start(task);
731
732   /* Prepare the task to send */
733   simdata_task_t t_simdata = task->simdata;
734   t_simdata->sender        = MSG_process_self();
735   t_simdata->source = MSG_host_self();
736   t_simdata->set_used();
737
738   msg_global->sent_msg++;
739
740   /* Try to send it */
741   try {
742     simgrid::s4u::CommPtr comm =
743         simgrid::s4u::Mailbox::by_name(alias)->put_init(task, t_simdata->bytes_amount)->set_rate(t_simdata->rate);
744     t_simdata->comm = comm;
745     comm->start();
746     if (TRACE_is_enabled() && task->category != nullptr)
747       simgrid::simix::simcall([comm, task] { comm->get_impl()->set_category(task->category); });
748     comm->wait_for(timeout);
749   } catch (simgrid::TimeoutError& e) {
750     ret = MSG_TIMEOUT;
751   } catch (simgrid::CancelException& e) {
752     ret = MSG_HOST_FAILURE;
753   } catch (xbt_ex& e) {
754     if (e.category == network_error)
755       ret = MSG_TRANSFER_FAILURE;
756     else
757       throw;
758
759     /* If the send failed, it is not used anymore */
760     t_simdata->set_not_used();
761   }
762
763   return ret;
764 }
765
766 /**
767  * @brief Sends a task to a mailbox with a timeout and with a maximum rate
768  *
769  * This is a blocking function, the execution flow will be blocked until the task is sent or the timeout is achieved.
770  *
771  * The maxrate parameter can be used to send a task with a limited
772  * bandwidth (smaller than the physical available value). Use
773  * MSG_task_send_with_timeout() if you don't limit the rate (or pass -1 as a rate
774  * value do disable this feature).
775  *
776  * @param task the task to be sent
777  * @param alias the mailbox name to where the task is sent
778  * @param timeout is the maximum wait time for completion (if -1, this call is the same as #MSG_task_send)
779  * @param maxrate the maximum communication rate for sending this task (byte/sec)
780  *
781  * @return Returns #MSG_OK if the task was successfully sent,
782  * #MSG_HOST_FAILURE, or #MSG_TRANSFER_FAILURE, or #MSG_TIMEOUT otherwise.
783  */
784 msg_error_t MSG_task_send_with_timeout_bounded(msg_task_t task, const char *alias, double timeout, double maxrate)
785 {
786   task->simdata->rate = maxrate;
787   return MSG_task_send_with_timeout(task, alias, timeout);
788 }
789
790 /**
791  * @brief Look if there is a communication on a mailbox and return the PID of the sender process.
792  *
793  * @param alias the name of the mailbox to be considered
794  *
795  * @return Returns the PID of sender process,
796  * -1 if there is no communication in the mailbox.
797  */
798 int MSG_task_listen_from(const char *alias)
799 {
800   simgrid::kernel::activity::CommImplPtr comm = simgrid::s4u::Mailbox::by_name(alias)->front();
801
802   return comm ? MSG_process_get_PID(static_cast<msg_task_t>(comm->src_buff_)->simdata->sender) : -1;
803 }
804
805 /**
806  * @brief Sets the tracing category of a task.
807  *
808  * This function should be called after the creation of a MSG task, to define the category of that task. The
809  * first parameter task must contain a task that was  created with the function #MSG_task_create. The second
810  * parameter category must contain a category that was previously declared with the function #TRACE_category
811  * (or with #TRACE_category_with_color).
812  *
813  * See @ref outcomes_vizu for details on how to trace the (categorized) resource utilization.
814  *
815  * @param task the task that is going to be categorized
816  * @param category the name of the category to be associated to the task
817  *
818  * @see MSG_task_get_category, TRACE_category, TRACE_category_with_color
819  */
820 void MSG_task_set_category (msg_task_t task, const char *category)
821 {
822   xbt_assert(task->category == nullptr, "Task %p(%s) already has a category (%s).", task, task->name, task->category);
823
824   // if user provides a nullptr category, task is no longer traced
825   if (category == nullptr) {
826     xbt_free(task->category);
827     task->category = nullptr;
828     XBT_DEBUG("MSG task %p(%s), category removed", task, task->name);
829   } else {
830     // set task category
831     task->category = xbt_strdup(category);
832     XBT_DEBUG("MSG task %p(%s), category %s", task, task->name, task->category);
833   }
834 }
835
836 /**
837  * @brief Gets the current tracing category of a task.
838  *
839  * @param task the task to be considered
840  *
841  * @see MSG_task_set_category
842  *
843  * @return Returns the name of the tracing category of the given task, nullptr otherwise
844  */
845 const char *MSG_task_get_category (msg_task_t task)
846 {
847   return task->category;
848 }