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