src/msg/msg_gos.c

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