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