src/msg/gos.c

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