src/msg/msg_gos.c

   1 /* Copyright (c) 2004-2011. 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 "msg_private.h"
   7 #include "msg_mailbox.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_gos_functions
  16  * \brief Executes a task and waits for its termination.
  17  *
  18  * This function is used for describing the behavior of an agent. It
  19  * takes only one parameter.
  20  * \param task a #m_task_t to execute on the location on which the
  21  agent is running.
  22  * \return #MSG_OK if the task was successfully completed, #MSG_TASK_CANCELED
  23  * or #MSG_HOST_FAILURE otherwise
  24  */
  25 MSG_error_t MSG_task_execute(m_task_t task)
  26 {
  27   simdata_task_t simdata = NULL;
  28   simdata_process_t p_simdata;
  29   e_smx_state_t comp_state;
  30   CHECK_HOST();
  31
  32   simdata = task->simdata;
  33
  34   xbt_assert(simdata->host_nb == 0,
  35               "This is a parallel task. Go to hell.");
  36
  37 #ifdef HAVE_TRACING
  38   TRACE_msg_task_execute_start(task);
  39 #endif
  40
  41   xbt_assert((!simdata->compute) && (task->simdata->isused == 0),
  42               "This task is executed somewhere else. Go fix your code! %d",
  43               task->simdata->isused);
  44
  45   XBT_DEBUG("Computing on %s", MSG_process_get_name(MSG_process_self()));
  46
  47   if (simdata->computation_amount == 0) {
  48 #ifdef HAVE_TRACING
  49     TRACE_msg_task_execute_end(task);
  50 #endif
  51     return MSG_OK;
  52   }
  53
  54   m_process_t self = SIMIX_process_self();
  55   p_simdata = SIMIX_process_self_get_data(self);
  56   simdata->isused=1;
  57   simdata->compute =
  58       SIMIX_req_host_execute(task->name, p_simdata->m_host->simdata->smx_host,
  59                            simdata->computation_amount,
  60                            simdata->priority);
  61 #ifdef HAVE_TRACING
  62   SIMIX_req_set_category(simdata->compute, task->category);
  63 #endif
  64
  65   p_simdata->waiting_action = simdata->compute;
  66   comp_state = SIMIX_req_host_execution_wait(simdata->compute);
  67   p_simdata->waiting_action = NULL;
  68
  69   simdata->isused=0;
  70
  71   XBT_DEBUG("Execution task '%s' finished in state %d", task->name, comp_state);
  72   if (comp_state == SIMIX_DONE) {
  73     /* action ended, set comm and compute = NULL, the actions is already destroyed in the main function */
  74     simdata->computation_amount = 0.0;
  75     simdata->comm = NULL;
  76     simdata->compute = NULL;
  77 #ifdef HAVE_TRACING
  78     TRACE_msg_task_execute_end(task);
  79 #endif
  80     MSG_RETURN(MSG_OK);
  81   } else if (SIMIX_req_host_get_state(SIMIX_host_self()) == 0) {
  82     /* action ended, set comm and compute = NULL, the actions is already destroyed in the main function */
  83     simdata->comm = NULL;
  84     simdata->compute = NULL;
  85 #ifdef HAVE_TRACING
  86     TRACE_msg_task_execute_end(task);
  87 #endif
  88     MSG_RETURN(MSG_HOST_FAILURE);
  89   } else {
  90     /* action ended, set comm and compute = NULL, the actions is already destroyed in the main function */
  91     simdata->comm = NULL;
  92     simdata->compute = NULL;
  93 #ifdef HAVE_TRACING
  94     TRACE_msg_task_execute_end(task);
  95 #endif
  96     MSG_RETURN(MSG_TASK_CANCELED);
  97   }
  98 }
  99
 100 /** \ingroup m_task_management
 101  * \brief Creates a new #m_task_t (a parallel one....).
 102  *
 103  * A constructor for #m_task_t taking six arguments and returning the
 104  corresponding object.
 105  * \param name a name for the object. It is for user-level information
 106  and can be NULL.
 107  * \param host_nb the number of hosts implied in the parallel task.
 108  * \param host_list an array of \p host_nb m_host_t.
 109  * \param computation_amount an array of \p host_nb
 110  doubles. computation_amount[i] is the total number of operations
 111  that have to be performed on host_list[i].
 112  * \param communication_amount an array of \p host_nb* \p host_nb doubles.
 113  * \param data a pointer to any data may want to attach to the new
 114  object.  It is for user-level information and can be NULL. It can
 115  be retrieved with the function \ref MSG_task_get_data.
 116  * \see m_task_t
 117  * \return The new corresponding object.
 118  */
 119 m_task_t
 120 MSG_parallel_task_create(const char *name, int host_nb,
 121                          const m_host_t * host_list,
 122                          double *computation_amount,
 123                          double *communication_amount, void *data)
 124 {
 125   int i;
 126   simdata_task_t simdata = xbt_new0(s_simdata_task_t, 1);
 127   m_task_t task = xbt_new0(s_m_task_t, 1);
 128   task->simdata = simdata;
 129
 130   /* Task structure */
 131   task->name = xbt_strdup(name);
 132   task->data = data;
 133
 134   /* Simulator Data */
 135   simdata->computation_amount = 0;
 136   simdata->message_size = 0;
 137   simdata->compute = NULL;
 138   simdata->comm = NULL;
 139   simdata->rate = -1.0;
 140   simdata->isused = 0;
 141   simdata->sender = NULL;
 142   simdata->receiver = NULL;
 143   simdata->source = NULL;
 144
 145   simdata->host_nb = host_nb;
 146   simdata->host_list = xbt_new0(smx_host_t, host_nb);
 147   simdata->comp_amount = computation_amount;
 148   simdata->comm_amount = communication_amount;
 149
 150   for (i = 0; i < host_nb; i++)
 151     simdata->host_list[i] = host_list[i]->simdata->smx_host;
 152
 153   return task;
 154 }
 155
 156 MSG_error_t MSG_parallel_task_execute(m_task_t task)
 157 {
 158   simdata_task_t simdata = NULL;
 159   e_smx_state_t comp_state;
 160   simdata_process_t p_simdata;
 161   CHECK_HOST();
 162
 163   simdata = task->simdata;
 164   p_simdata = SIMIX_process_self_get_data(SIMIX_process_self());
 165
 166   xbt_assert((!simdata->compute)
 167               && (task->simdata->isused == 0),
 168               "This task is executed somewhere else. Go fix your code!");
 169
 170   xbt_assert(simdata->host_nb,
 171               "This is not a parallel task. Go to hell.");
 172
 173   XBT_DEBUG("Parallel computing on %s", p_simdata->m_host->name);
 174
 175   simdata->isused=1;
 176
 177   simdata->compute =
 178       SIMIX_req_host_parallel_execute(task->name, simdata->host_nb,
 179                                   simdata->host_list,
 180                                   simdata->comp_amount,
 181                                   simdata->comm_amount, 1.0, -1.0);
 182   XBT_DEBUG("Parallel execution action created: %p", simdata->compute);
 183
 184   p_simdata->waiting_action = simdata->compute;
 185   comp_state = SIMIX_req_host_execution_wait(simdata->compute);
 186   p_simdata->waiting_action = NULL;
 187
 188   XBT_DEBUG("Finished waiting for execution of action %p, state = %d", simdata->compute, comp_state);
 189
 190   simdata->isused=0;
 191
 192   if (comp_state == SIMIX_DONE) {
 193     /* action ended, set comm and compute = NULL, the actions is already destroyed in the main function */
 194     simdata->computation_amount = 0.0;
 195     simdata->comm = NULL;
 196     simdata->compute = NULL;
 197     MSG_RETURN(MSG_OK);
 198   } else if (SIMIX_req_host_get_state(SIMIX_host_self()) == 0) {
 199     /* action ended, set comm and compute = NULL, the actions is already destroyed in the main function */
 200     simdata->comm = NULL;
 201     simdata->compute = NULL;
 202     MSG_RETURN(MSG_HOST_FAILURE);
 203   } else {
 204     /* action ended, set comm and compute = NULL, the actions is already destroyed in the main function */
 205     simdata->comm = NULL;
 206     simdata->compute = NULL;
 207     MSG_RETURN(MSG_TASK_CANCELED);
 208   }
 209 }
 210
 211
 212 /** \ingroup msg_gos_functions
 213  * \brief Sleep for the specified number of seconds
 214  *
 215  * Makes the current process sleep until \a time seconds have elapsed.
 216  *
 217  * \param nb_sec a number of second
 218  */
 219 MSG_error_t MSG_process_sleep(double nb_sec)
 220 {
 221   e_smx_state_t state;
 222   /*m_process_t proc = MSG_process_self();*/
 223
 224 #ifdef HAVE_TRACING
 225   TRACE_msg_process_sleep_in(MSG_process_self());
 226 #endif
 227
 228   /* create action to sleep */
 229   state = SIMIX_req_process_sleep(nb_sec);
 230
 231   /*proc->simdata->waiting_action = act_sleep;
 232
 233   FIXME: check if not setting the waiting_action breaks something on msg
 234
 235   proc->simdata->waiting_action = NULL;*/
 236
 237   if (state == SIMIX_DONE) {
 238 #ifdef HAVE_TRACING
 239   TRACE_msg_process_sleep_out(MSG_process_self());
 240 #endif
 241     MSG_RETURN(MSG_OK);
 242   } else {
 243 #ifdef HAVE_TRACING
 244     TRACE_msg_process_sleep_out(MSG_process_self());
 245 #endif
 246     MSG_RETURN(MSG_HOST_FAILURE);
 247   }
 248 }
 249
 250 /** \ingroup msg_gos_functions
 251  * \brief Listen on \a channel and waits for receiving a task from \a host.
 252  *
 253  * It takes three parameters.
 254  * \param task a memory location for storing a #m_task_t. It will
 255  hold a task when this function will return. Thus \a task should not
 256  be equal to \c NULL and \a *task should be equal to \c NULL. If one of
 257  those two condition does not hold, there will be a warning message.
 258  * \param channel the channel on which the agent should be
 259  listening. This value has to be >=0 and < than the maximal
 260  number of channels fixed with MSG_set_channel_number().
 261  * \param host the host that is to be watched.
 262  * \return a #MSG_error_t indicating whether the operation was successful (#MSG_OK), or why it failed otherwise.
 263  */
 264 MSG_error_t
 265 MSG_task_get_from_host(m_task_t * task, m_channel_t channel, m_host_t host)
 266 {
 267   return MSG_task_get_ext(task, channel, -1, host);
 268 }
 269
 270 /** \ingroup msg_gos_functions
 271  * \brief Listen on a channel and wait for receiving a task.
 272  *
 273  * It takes two parameters.
 274  * \param task a memory location for storing a #m_task_t. It will
 275  hold a task when this function will return. Thus \a task should not
 276  be equal to \c NULL and \a *task should be equal to \c NULL. If one of
 277  those two condition does not hold, there will be a warning message.
 278  * \param channel the channel on which the agent should be
 279  listening. This value has to be >=0 and < than the maximal
 280  number of channels fixed with MSG_set_channel_number().
 281  * \return a #MSG_error_t indicating whether the operation was successful (#MSG_OK), or why it failed otherwise.
 282  */
 283 MSG_error_t MSG_task_get(m_task_t * task, m_channel_t channel)
 284 {
 285   return MSG_task_get_with_timeout(task, channel, -1);
 286 }
 287
 288 /** \ingroup msg_gos_functions
 289  * \brief Listen on a channel and wait for receiving a task with a timeout.
 290  *
 291  * It takes three parameters.
 292  * \param task a memory location for storing a #m_task_t. It will
 293  hold a task when this function will return. Thus \a task should not
 294  be equal to \c NULL and \a *task should be equal to \c NULL. If one of
 295  those two condition does not hold, there will be a warning message.
 296  * \param channel the channel on which the agent should be
 297  listening. This value has to be >=0 and < than the maximal
 298  number of channels fixed with MSG_set_channel_number().
 299  * \param max_duration the maximum time to wait for a task before giving
 300  up. In such a case, #MSG_TRANSFER_FAILURE will be returned, \a task
 301  will not be modified and will still be
 302  equal to \c NULL when returning.
 303  * \return a #MSG_error_t indicating whether the operation was successful (#MSG_OK), or why it failed otherwise.
 304  */
 305 MSG_error_t
 306 MSG_task_get_with_timeout(m_task_t * task, m_channel_t channel,
 307                           double max_duration)
 308 {
 309   return MSG_task_get_ext(task, channel, max_duration, NULL);
 310 }
 311
 312 /** \defgroup msg_gos_functions MSG Operating System Functions
 313  *  \brief This section describes the functions that can be used
 314  *  by an agent for handling some task.
 315  */
 316
 317 MSG_error_t
 318 MSG_task_get_ext(m_task_t * task, m_channel_t channel, double timeout,
 319                  m_host_t host)
 320 {
 321   xbt_assert((channel >= 0)
 322               && (channel < msg_global->max_channel), "Invalid channel %d",
 323               channel);
 324
 325   return
 326       MSG_mailbox_get_task_ext(MSG_mailbox_get_by_channel
 327                                (MSG_host_self(), channel), task, host,
 328                                timeout);
 329 }
 330
 331 MSG_error_t
 332 MSG_task_receive_from_host(m_task_t * task, const char *alias,
 333                            m_host_t host)
 334 {
 335   return MSG_task_receive_ext(task, alias, -1, host);
 336 }
 337
 338 MSG_error_t MSG_task_receive(m_task_t * task, const char *alias)
 339 {
 340   return MSG_task_receive_with_timeout(task, alias, -1);
 341 }
 342
 343 MSG_error_t
 344 MSG_task_receive_with_timeout(m_task_t * task, const char *alias,
 345                               double timeout)
 346 {
 347   return MSG_task_receive_ext(task, alias, timeout, NULL);
 348 }
 349
 350 MSG_error_t
 351 MSG_task_receive_ext(m_task_t * task, const char *alias, double timeout,
 352                      m_host_t host)
 353 {
 354   XBT_DEBUG
 355       ("MSG_task_receive_ext: Trying to receive a message on mailbox '%s'",
 356        alias);
 357   return MSG_mailbox_get_task_ext(MSG_mailbox_get_by_alias(alias), task,
 358                                   host, timeout);
 359 }
 360
 361 /** \ingroup msg_gos_functions
 362  * \brief Sends a task on a mailbox.
 363  *
 364  * This is a non blocking function: use MSG_comm_wait() or MSG_comm_test()
 365  * to end the communication.
 366  *
 367  * \param task a #m_task_t to send on another location.
 368  * \param alias name of the mailbox to sent the task to
 369  * \return the msg_comm_t communication created
 370  */
 371 msg_comm_t MSG_task_isend(m_task_t task, const char *alias)
 372 {
 373   return MSG_task_isend_with_matching(task,alias,NULL,NULL);
 374 }
 375
 376 /** \ingroup msg_gos_functions
 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()
 380  * to end the communication.
 381  *
 382  * \param task a #m_task_t to send on another location.
 383  * \param alias name of the mailbox to sent the task to
 384  * \param match_fun boolean function taking the #match_data provided by sender (here), and the one of the receiver (if any) and returning whether they match
 385  * \param match_data user provided data passed to match_fun
 386  * \return the msg_comm_t communication created
 387  */
 388 XBT_INLINE msg_comm_t MSG_task_isend_with_matching(m_task_t task, const char *alias,
 389     int (*match_fun)(void*,void*),
 390     void *match_data)
 391 {
 392   simdata_task_t t_simdata = NULL;
 393   m_process_t process = MSG_process_self();
 394   msg_mailbox_t mailbox = MSG_mailbox_get_by_alias(alias);
 395
 396   CHECK_HOST();
 397
 398   /* FIXME: these functions are not traceable */
 399
 400   /* Prepare the task to send */
 401   t_simdata = task->simdata;
 402   t_simdata->sender = process;
 403   t_simdata->source = ((simdata_process_t) SIMIX_process_self_get_data(process))->m_host;
 404
 405   xbt_assert(t_simdata->isused == 0,
 406               "This task is still being used somewhere else. You cannot send it now. Go fix your code!");
 407
 408   t_simdata->isused = 1;
 409   msg_global->sent_msg++;
 410
 411   /* Send it by calling SIMIX network layer */
 412   msg_comm_t comm = xbt_new0(s_msg_comm_t, 1);
 413   comm->task_sent = task;
 414   comm->task_received = NULL;
 415   comm->status = MSG_OK;
 416   comm->s_comm =
 417     SIMIX_req_comm_isend(mailbox, t_simdata->message_size,
 418                          t_simdata->rate, task, sizeof(void *), match_fun, NULL, match_data, 0);
 419   t_simdata->comm = comm->s_comm; /* FIXME: is the field t_simdata->comm still useful? */
 420
 421   return comm;
 422 }
 423
 424 /** \ingroup msg_gos_functions
 425  * \brief Sends a task on a mailbox.
 426  *
 427  * This is a non blocking detached send function.
 428  * Think of it as a best effort send. Keep in mind that the third parameter
 429  * is only called if the communication fails. If the communication does work,
 430  * it is responsibility of the receiver code to free anything related to
 431  * the task, as usual. More details on this can be obtained on
 432  * <a href="http://lists.gforge.inria.fr/pipermail/simgrid-user/2011-November/002649.html">this thread</a>
 433  * in the SimGrid-user mailing list archive.
 434  *
 435  * \param task a #m_task_t to send on another location.
 436  * \param alias name of the mailbox to sent the task to
 437  * \param cleanup a function to destroy the task if the
 438  * communication fails, e.g. MSG_task_destroy
 439  * (if NULL, no function will be called)
 440  */
 441 void MSG_task_dsend(m_task_t task, const char *alias, void_f_pvoid_t cleanup)
 442 {
 443   simdata_task_t t_simdata = NULL;
 444   m_process_t process = MSG_process_self();
 445   msg_mailbox_t mailbox = MSG_mailbox_get_by_alias(alias);
 446
 447   CHECK_HOST();
 448
 449   /* FIXME: these functions are not traceable */
 450
 451   /* Prepare the task to send */
 452   t_simdata = task->simdata;
 453   t_simdata->sender = process;
 454   t_simdata->source = ((simdata_process_t) SIMIX_process_self_get_data(process))->m_host;
 455
 456   xbt_assert(t_simdata->isused == 0,
 457               "This task is still being used somewhere else. You cannot send it now. Go fix your code!");
 458
 459   t_simdata->isused = 1;
 460   msg_global->sent_msg++;
 461
 462   /* Send it by calling SIMIX network layer */
 463   smx_action_t comm = SIMIX_req_comm_isend(mailbox, t_simdata->message_size,
 464                        t_simdata->rate, task, sizeof(void *), NULL, cleanup, NULL, 1);
 465   t_simdata->comm = comm;
 466 }
 467
 468 /** \ingroup msg_gos_functions
 469  * \brief Starts listening for receiving a task from an asynchronous communication.
 470  *
 471  * This is a non blocking function: use MSG_comm_wait() or MSG_comm_test()
 472  * to end the communication.
 473  *
 474  * \param task a memory location for storing a #m_task_t.
 475  * \param name of the mailbox to receive the task on
 476  * \return the msg_comm_t communication created
 477  */
 478 msg_comm_t MSG_task_irecv(m_task_t *task, const char *name)
 479 {
 480   smx_rdv_t rdv = MSG_mailbox_get_by_alias(name);
 481
 482   CHECK_HOST();
 483
 484   /* FIXME: these functions are not tracable */
 485
 486   /* Sanity check */
 487   xbt_assert(task, "Null pointer for the task storage");
 488
 489   if (*task)
 490     XBT_CRITICAL
 491         ("MSG_task_irecv() was asked to write in a non empty task struct.");
 492
 493   /* Try to receive it by calling SIMIX network layer */
 494   msg_comm_t comm = xbt_new0(s_msg_comm_t, 1);
 495   comm->task_sent = NULL;
 496   comm->task_received = task;
 497   comm->status = MSG_OK;
 498   comm->s_comm = SIMIX_req_comm_irecv(rdv, task, NULL, NULL, NULL);
 499
 500   return comm;
 501 }
 502
 503 /** \ingroup msg_gos_functions
 504  * \brief Checks whether a communication is done, and if yes, finalizes it.
 505  * \param comm the communication to test
 506  * \return TRUE if the communication is finished
 507  * (but it may have failed, use MSG_comm_get_status() to know its status)
 508  * or FALSE if the communication is not finished yet
 509  * If the status is FALSE, don't forget to use MSG_process_sleep() after the test.
 510  */
 511 int MSG_comm_test(msg_comm_t comm)
 512 {
 513   xbt_ex_t e;
 514   int finished = 0;
 515   TRY {
 516     finished = SIMIX_req_comm_test(comm->s_comm);
 517
 518     if (finished && comm->task_received != NULL) {
 519       /* I am the receiver */
 520       (*comm->task_received)->simdata->isused = 0;
 521     }
 522   }
 523   CATCH(e) {
 524     switch (e.category) {
 525
 526       case host_error:
 527         comm->status = MSG_HOST_FAILURE;
 528         finished = 1;
 529         break;
 530
 531       case network_error:
 532         comm->status = MSG_TRANSFER_FAILURE;
 533         finished = 1;
 534         break;
 535
 536       case timeout_error:
 537         comm->status = MSG_TIMEOUT;
 538         finished = 1;
 539         break;
 540
 541       default:
 542         RETHROW;
 543     }
 544     xbt_ex_free(e);
 545   }
 546
 547   return finished;
 548 }
 549
 550 /** \ingroup msg_gos_functions
 551  * \brief This function checks if a communication is finished.
 552  * \param comms a vector of communications
 553  * \return the position of the finished communication if any
 554  * (but it may have failed, use MSG_comm_get_status() to know its status),
 555  * or -1 if none is finished
 556  */
 557 int MSG_comm_testany(xbt_dynar_t comms)
 558 {
 559   xbt_ex_t e;
 560   int finished_index = -1;
 561
 562   /* create the equivalent dynar with SIMIX objects */
 563   xbt_dynar_t s_comms = xbt_dynar_new(sizeof(smx_action_t), NULL);
 564   msg_comm_t comm;
 565   unsigned int cursor;
 566   xbt_dynar_foreach(comms, cursor, comm) {
 567     xbt_dynar_push(s_comms, &comm->s_comm);
 568   }
 569
 570   MSG_error_t status = MSG_OK;
 571   TRY {
 572     finished_index = SIMIX_req_comm_testany(s_comms);
 573   }
 574   CATCH(e) {
 575     switch (e.category) {
 576
 577       case host_error:
 578         finished_index = e.value;
 579         status = MSG_HOST_FAILURE;
 580         break;
 581
 582       case network_error:
 583         finished_index = e.value;
 584         status = MSG_TRANSFER_FAILURE;
 585         break;
 586
 587       case timeout_error:
 588         finished_index = e.value;
 589         status = MSG_TIMEOUT;
 590         break;
 591
 592       default:
 593         RETHROW;
 594     }
 595     xbt_ex_free(e);
 596   }
 597   xbt_dynar_free(&s_comms);
 598
 599   if (finished_index != -1) {
 600     comm = xbt_dynar_get_as(comms, finished_index, msg_comm_t);
 601     /* the communication is finished */
 602     comm->status = status;
 603
 604     if (status == MSG_OK && comm->task_received != NULL) {
 605       /* I am the receiver */
 606       (*comm->task_received)->simdata->isused = 0;
 607     }
 608   }
 609
 610   return finished_index;
 611 }
 612
 613 /** \ingroup msg_gos_functions
 614  * \brief Destroys a communication.
 615  * \param comm the communication to destroy.
 616  */
 617 void MSG_comm_destroy(msg_comm_t comm)
 618 {
 619   xbt_free(comm);
 620 }
 621
 622 /** \ingroup msg_gos_functions
 623  * \brief Wait for the completion of a communication.
 624  *
 625  * It takes two parameters.
 626  * \param comm the communication to wait.
 627  * \param timeout Wait until the communication terminates or the timeout occurs
 628  * \return MSG_error_t
 629  */
 630 MSG_error_t MSG_comm_wait(msg_comm_t comm, double timeout)
 631 {
 632   xbt_ex_t e;
 633   TRY {
 634     SIMIX_req_comm_wait(comm->s_comm, timeout);
 635
 636     if (comm->task_received != NULL) {
 637       /* I am the receiver */
 638       (*comm->task_received)->simdata->isused = 0;
 639     }
 640
 641     /* FIXME: these functions are not traceable */
 642   }
 643   CATCH(e) {
 644     switch (e.category) {
 645     case host_error:
 646       comm->status = MSG_HOST_FAILURE;
 647       break;
 648     case network_error:
 649       comm->status = MSG_TRANSFER_FAILURE;
 650       break;
 651     case timeout_error:
 652       comm->status = MSG_TIMEOUT;
 653       break;
 654     default:
 655       RETHROW;
 656     }
 657     xbt_ex_free(e);
 658   }
 659
 660   return comm->status;
 661 }
 662
 663 /** \ingroup msg_gos_functions
 664 * \brief This function is called by a sender and permit to wait for each communication
 665 *
 666 * \param comm a vector of communication
 667 * \param nb_elem is the size of the comm vector
 668 * \param timeout for each call of MSG_comm_wait
 669 */
 670 void MSG_comm_waitall(msg_comm_t * comm, int nb_elem, double timeout)
 671 {
 672   int i = 0;
 673   for (i = 0; i < nb_elem; i++) {
 674     MSG_comm_wait(comm[i], timeout);
 675   }
 676 }
 677
 678 /** \ingroup msg_gos_functions
 679  * \brief This function waits for the first communication finished in a list.
 680  * \param comms a vector of communications
 681  * \return the position of the first finished communication
 682  * (but it may have failed, use MSG_comm_get_status() to know its status)
 683  */
 684 int MSG_comm_waitany(xbt_dynar_t comms)
 685 {
 686   xbt_ex_t e;
 687   int finished_index = -1;
 688
 689   /* create the equivalent dynar with SIMIX objects */
 690   xbt_dynar_t s_comms = xbt_dynar_new(sizeof(smx_action_t), NULL);
 691   msg_comm_t comm;
 692   unsigned int cursor;
 693   xbt_dynar_foreach(comms, cursor, comm) {
 694     xbt_dynar_push(s_comms, &comm->s_comm);
 695   }
 696
 697   MSG_error_t status = MSG_OK;
 698   TRY {
 699     finished_index = SIMIX_req_comm_waitany(s_comms);
 700   }
 701   CATCH(e) {
 702     switch (e.category) {
 703
 704       case host_error:
 705         finished_index = e.value;
 706         status = MSG_HOST_FAILURE;
 707         break;
 708
 709       case network_error:
 710         finished_index = e.value;
 711         status = MSG_TRANSFER_FAILURE;
 712         break;
 713
 714       case timeout_error:
 715         finished_index = e.value;
 716         status = MSG_TIMEOUT;
 717         break;
 718
 719       default:
 720         RETHROW;
 721     }
 722     xbt_ex_free(e);
 723   }
 724
 725   xbt_assert(finished_index != -1, "WaitAny returned -1");
 726   xbt_dynar_free(&s_comms);
 727
 728   comm = xbt_dynar_get_as(comms, finished_index, msg_comm_t);
 729   /* the communication is finished */
 730   comm->status = status;
 731
 732   if (comm->task_received != NULL) {
 733     /* I am the receiver */
 734     (*comm->task_received)->simdata->isused = 0;
 735   }
 736
 737   return finished_index;
 738 }
 739
 740 /**
 741  * \ingroup msg_gos_functions
 742  * \brief Returns the error (if any) that occured during a finished communication.
 743  * \param comm a finished communication
 744  * \return the status of the communication, or MSG_OK if no error occured
 745  * during the communication
 746  */
 747 MSG_error_t MSG_comm_get_status(msg_comm_t comm) {
 748
 749   return comm->status;
 750 }
 751
 752 m_task_t MSG_comm_get_task(msg_comm_t comm)
 753 {
 754   xbt_assert(comm, "Invalid parameter");
 755
 756   return comm->task_received ? *comm->task_received : comm->task_sent;
 757 }
 758
 759 /**
 760  * \brief This function is called by SIMIX to copy the data of a comm.
 761  * \param comm the comm
 762  * \param buff the data copied
 763  * \param buff_size size of the buffer
 764  */
 765 void MSG_comm_copy_data_from_SIMIX(smx_action_t comm, void* buff, size_t buff_size) {
 766
 767   // copy the task
 768   SIMIX_comm_copy_pointer_callback(comm, buff, buff_size);
 769
 770   // notify the user callback if any
 771   if (msg_global->task_copy_callback) {
 772     m_task_t task = buff;
 773     msg_global->task_copy_callback(task,
 774         SIMIX_req_comm_get_src_proc(comm), SIMIX_req_comm_get_dst_proc(comm));
 775   }
 776 }
 777
 778 /** \ingroup msg_gos_functions
 779  * \brief Put a task on a channel of an host and waits for the end of the
 780  * transmission.
 781  *
 782  * This function is used for describing the behavior of an agent. It
 783  * takes three parameter.
 784  * \param task a #m_task_t to send on another location. This task
 785  will not be usable anymore when the function will return. There is
 786  no automatic task duplication and you have to save your parameters
 787  before calling this function. Tasks are unique and once it has been
 788  sent to another location, you should not access it anymore. You do
 789  not need to call MSG_task_destroy() but to avoid using, as an
 790  effect of inattention, this task anymore, you definitely should
 791  renitialize it with #MSG_TASK_UNINITIALIZED. Note that this task
 792  can be transfered iff it has been correctly created with
 793  MSG_task_create().
 794  * \param dest the destination of the message
 795  * \param channel the channel on which the agent should put this
 796  task. This value has to be >=0 and < than the maximal number of
 797  channels fixed with MSG_set_channel_number().
 798  * \return #MSG_HOST_FAILURE if the host on which
 799  * this function was called was shut down,
 800  * #MSG_TRANSFER_FAILURE if the transfer could not be properly done
 801  * (network failure, dest failure) or #MSG_OK if it succeeded.
 802  */
 803 MSG_error_t MSG_task_put(m_task_t task, m_host_t dest, m_channel_t channel)
 804 {
 805   return MSG_task_put_with_timeout(task, dest, channel, -1.0);
 806 }
 807
 808 /** \ingroup msg_gos_functions
 809  * \brief Does exactly the same as MSG_task_put but with a bounded transmition
 810  * rate.
 811  *
 812  * \sa MSG_task_put
 813  */
 814 MSG_error_t
 815 MSG_task_put_bounded(m_task_t task, m_host_t dest, m_channel_t channel,
 816                      double maxrate)
 817 {
 818   task->simdata->rate = maxrate;
 819   return MSG_task_put(task, dest, channel);
 820 }
 821
 822 /** \ingroup msg_gos_functions \brief Put a task on a channel of an
 823  * host (with a timeout on the waiting of the destination host) and
 824  * waits for the end of the transmission.
 825  *
 826  * This function is used for describing the behavior of an agent. It
 827  * takes four parameter.
 828  * \param task a #m_task_t to send on another location. This task
 829  will not be usable anymore when the function will return. There is
 830  no automatic task duplication and you have to save your parameters
 831  before calling this function. Tasks are unique and once it has been
 832  sent to another location, you should not access it anymore. You do
 833  not need to call MSG_task_destroy() but to avoid using, as an
 834  effect of inattention, this task anymore, you definitely should
 835  renitialize it with #MSG_TASK_UNINITIALIZED. Note that this task
 836  can be transfered iff it has been correctly created with
 837  MSG_task_create().
 838  * \param dest the destination of the message
 839  * \param channel the channel on which the agent should put this
 840  task. This value has to be >=0 and < than the maximal number of
 841  channels fixed with MSG_set_channel_number().
 842  * \param timeout the maximum time to wait for a task before giving
 843  up. In such a case, #MSG_TRANSFER_FAILURE will be returned, \a task
 844  will not be modified
 845  * \return #MSG_HOST_FAILURE if the host on which
 846 this function was called was shut down,
 847 #MSG_TRANSFER_FAILURE if the transfer could not be properly done
 848 (network failure, dest failure, timeout...) or #MSG_OK if the communication succeeded.
 849  */
 850 MSG_error_t
 851 MSG_task_put_with_timeout(m_task_t task, m_host_t dest,
 852                           m_channel_t channel, double timeout)
 853 {
 854   xbt_assert((channel >= 0)
 855               && (channel < msg_global->max_channel), "Invalid channel %d",
 856               channel);
 857
 858   XBT_DEBUG("MSG_task_put_with_timout: Trying to send a task to '%s'", dest->name);
 859   return
 860       MSG_mailbox_put_with_timeout(MSG_mailbox_get_by_channel
 861                                    (dest, channel), task, timeout);
 862 }
 863
 864 MSG_error_t MSG_task_send(m_task_t task, const char *alias)
 865 {
 866   XBT_DEBUG("MSG_task_send: Trying to send a message on mailbox '%s'", alias);
 867   return MSG_task_send_with_timeout(task, alias, -1);
 868 }
 869
 870
 871 MSG_error_t
 872 MSG_task_send_bounded(m_task_t task, const char *alias, double maxrate)
 873 {
 874   task->simdata->rate = maxrate;
 875   return MSG_task_send(task, alias);
 876 }
 877
 878
 879 MSG_error_t
 880 MSG_task_send_with_timeout(m_task_t task, const char *alias,
 881                            double timeout)
 882 {
 883   return MSG_mailbox_put_with_timeout(MSG_mailbox_get_by_alias(alias),
 884                                       task, timeout);
 885 }
 886
 887 int MSG_task_listen(const char *alias)
 888 {
 889   CHECK_HOST();
 890
 891   return !MSG_mailbox_is_empty(MSG_mailbox_get_by_alias(alias));
 892 }
 893
 894 /** \ingroup msg_gos_functions
 895  * \brief Test whether there is a pending communication on a channel.
 896  *
 897  * It takes one parameter.
 898  * \param channel the channel on which the agent should be
 899  listening. This value has to be >=0 and < than the maximal
 900  number of channels fixed with MSG_set_channel_number().
 901  * \return 1 if there is a pending communication and 0 otherwise
 902  */
 903 int MSG_task_Iprobe(m_channel_t channel)
 904 {
 905   xbt_assert((channel >= 0)
 906               && (channel < msg_global->max_channel), "Invalid channel %d",
 907               channel);
 908
 909   CHECK_HOST();
 910
 911   return
 912       !MSG_mailbox_is_empty(MSG_mailbox_get_by_channel
 913                             (MSG_host_self(), channel));
 914 }
 915
 916 /** \ingroup msg_gos_functions
 917
 918  * \brief Return the number of tasks waiting to be received on a \a
 919  channel and sent by \a host.
 920  *
 921  * It takes two parameters.
 922  * \param channel the channel on which the agent should be
 923  listening. This value has to be >=0 and < than the maximal
 924  number of channels fixed with MSG_set_channel_number().
 925  * \param host the host that is to be watched.
 926  * \return the number of tasks waiting to be received on \a channel
 927  and sent by \a host.
 928  */
 929 int MSG_task_probe_from_host(int channel, m_host_t host)
 930 {
 931   xbt_assert((channel >= 0)
 932               && (channel < msg_global->max_channel), "Invalid channel %d",
 933               channel);
 934
 935   CHECK_HOST();
 936
 937   return
 938       MSG_mailbox_get_count_host_waiting_tasks(MSG_mailbox_get_by_channel
 939                                                (MSG_host_self(), channel),
 940                                                host);
 941
 942 }
 943
 944 int MSG_task_listen_from_host(const char *alias, m_host_t host)
 945 {
 946   CHECK_HOST();
 947
 948   return
 949       MSG_mailbox_get_count_host_waiting_tasks(MSG_mailbox_get_by_alias
 950                                                (alias), host);
 951 }
 952
 953 /** \ingroup msg_gos_functions
 954  * \brief Test whether there is a pending communication on a channel, and who sent it.
 955  *
 956  * It takes one parameter.
 957  * \param channel the channel on which the agent should be
 958  listening. This value has to be >=0 and < than the maximal
 959  number of channels fixed with MSG_set_channel_number().
 960  * \return -1 if there is no pending communication and the PID of the process who sent it otherwise
 961  */
 962 int MSG_task_probe_from(m_channel_t channel)
 963 {
 964   m_task_t task;
 965
 966   CHECK_HOST();
 967
 968   xbt_assert((channel >= 0)
 969               && (channel < msg_global->max_channel), "Invalid channel %d",
 970               channel);
 971
 972   if (NULL ==
 973       (task =
 974        MSG_mailbox_get_head(MSG_mailbox_get_by_channel
 975                             (MSG_host_self(), channel))))
 976     return -1;
 977
 978   return MSG_process_get_PID(task->simdata->sender);
 979 }
 980
 981 int MSG_task_listen_from(const char *alias)
 982 {
 983   m_task_t task;
 984
 985   CHECK_HOST();
 986
 987   if (NULL ==
 988       (task = MSG_mailbox_get_head(MSG_mailbox_get_by_alias(alias))))
 989     return -1;
 990
 991   return MSG_process_get_PID(task->simdata->sender);
 992 }
 993
 994 #ifdef MSG_USE_DEPRECATED
 995 /** \ingroup msg_gos_functions
 996  *
 997  * \brief Return the last value returned by a MSG function (except
 998  * MSG_get_errno...).
 999  */
1000 MSG_error_t MSG_get_errno(void)
1001 {
1002   return PROCESS_GET_ERRNO();
1003 }
1004 #endif