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