src/msg/gos.c

   1 /*     $Id$      */
   2
   3 /* Copyright (c) 2002-2007 Arnaud Legrand.                                  */
   4 /* Copyright (c) 2007 Bruno Donassolo.                                      */
   5 /* All rights reserved.                                                     */
   6
   7 /* This program is free software; you can redistribute it and/or modify it
   8  * under the terms of the license (GNU LGPL) which comes with this package. */
   9
  10 #include "msg/private.h"
  11 #include "xbt/sysdep.h"
  12 #include "xbt/log.h"
  13 #include "mailbox.h"
  14
  15
  16 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(msg_gos, msg,
  17                                 "Logging specific to MSG (gos)");
  18
  19 /** \ingroup msg_gos_functions
  20  *
  21  * \brief Return the last value returned by a MSG function (except
  22  * MSG_get_errno...).
  23  */
  24 MSG_error_t MSG_get_errno(void)
  25 {
  26   return PROCESS_GET_ERRNO();
  27 }
  28
  29 /** \ingroup msg_gos_functions
  30  * \brief Executes a task and waits for its termination.
  31  *
  32  * This function is used for describing the behavior of an agent. It
  33  * takes only one parameter.
  34  * \param task a #m_task_t to execute on the location on which the
  35    agent is running.
  36  * \return #MSG_FATAL if \a task is not properly initialized and
  37  * #MSG_OK otherwise.
  38  */
  39 MSG_error_t MSG_task_execute(m_task_t task)
  40 {
  41   simdata_task_t simdata = NULL;
  42   m_process_t self = MSG_process_self();
  43   e_surf_action_state_t state = SURF_ACTION_NOT_IN_THE_SYSTEM;
  44   CHECK_HOST();
  45 #ifdef HAVE_TRACING
  46   TRACE_msg_task_execute_start (task);
  47 #endif
  48
  49   simdata = task->simdata;
  50
  51   xbt_assert1((!simdata->compute) && (task->simdata->refcount == 1),
  52               "This task is executed somewhere else. Go fix your code! %d", task->simdata->refcount);
  53
  54   DEBUG1("Computing on %s", MSG_process_self()->simdata->m_host->name);
  55
  56   simdata->refcount++;
  57   SIMIX_mutex_lock(simdata->mutex);
  58   simdata->compute =
  59     SIMIX_action_execute(SIMIX_host_self(), task->name,
  60                          simdata->computation_amount);
  61   SIMIX_action_set_priority(simdata->compute, simdata->priority);
  62
  63   /* changed to waiting action since we are always waiting one action (execute, communicate or sleep) */
  64   self->simdata->waiting_action = simdata->compute;
  65   SIMIX_register_action_to_condition(simdata->compute, simdata->cond);
  66   do {
  67     SIMIX_cond_wait(simdata->cond, simdata->mutex);
  68     state = SIMIX_action_get_state(simdata->compute);
  69   } while (state == SURF_ACTION_READY || state == SURF_ACTION_RUNNING);
  70   SIMIX_unregister_action_to_condition(simdata->compute, simdata->cond);
  71   self->simdata->waiting_action = NULL;
  72
  73   SIMIX_mutex_unlock(simdata->mutex);
  74   simdata->refcount--;
  75
  76   if (SIMIX_action_get_state(task->simdata->compute) == SURF_ACTION_DONE) {
  77     /* action ended, set comm and compute = NULL, the actions is already destroyed in the main function */
  78     SIMIX_action_destroy(task->simdata->compute);
  79     simdata->computation_amount = 0.0;
  80     simdata->comm = NULL;
  81     simdata->compute = NULL;
  82 #ifdef HAVE_TRACING
  83     TRACE_msg_task_execute_end (task);
  84 #endif
  85     MSG_RETURN(MSG_OK);
  86   } else if (SIMIX_host_get_state(SIMIX_host_self()) == 0) {
  87     /* action ended, set comm and compute = NULL, the actions is already destroyed in the main function */
  88     SIMIX_action_destroy(task->simdata->compute);
  89     simdata->comm = NULL;
  90     simdata->compute = NULL;
  91 #ifdef HAVE_TRACING
  92     TRACE_msg_task_execute_end (task);
  93 #endif
  94     MSG_RETURN(MSG_HOST_FAILURE);
  95   } else {
  96     /* action ended, set comm and compute = NULL, the actions is already destroyed in the main function */
  97     SIMIX_action_destroy(task->simdata->compute);
  98     simdata->comm = NULL;
  99     simdata->compute = NULL;
 100 #ifdef HAVE_TRACING
 101     TRACE_msg_task_execute_end (task);
 102 #endif
 103     MSG_RETURN(MSG_TASK_CANCELLED);
 104   }
 105 }
 106
 107 /** \ingroup m_task_management
 108  * \brief Creates a new #m_task_t (a parallel one....).
 109  *
 110  * A constructor for #m_task_t taking six arguments and returning the
 111    corresponding object.
 112  * \param name a name for the object. It is for user-level information
 113    and can be NULL.
 114  * \param host_nb the number of hosts implied in the parallel task.
 115  * \param host_list an array of \p host_nb m_host_t.
 116  * \param computation_amount an array of \p host_nb
 117    doubles. computation_amount[i] is the total number of operations
 118    that have to be performed on host_list[i].
 119  * \param communication_amount an array of \p host_nb* \p host_nb doubles.
 120  * \param data a pointer to any data may want to attach to the new
 121    object.  It is for user-level information and can be NULL. It can
 122    be retrieved with the function \ref MSG_task_get_data.
 123  * \see m_task_t
 124  * \return The new corresponding object.
 125  */
 126 m_task_t
 127 MSG_parallel_task_create(const char *name, int host_nb,
 128                          const m_host_t * host_list,
 129                          double *computation_amount,
 130                          double *communication_amount, void *data)
 131 {
 132   int i;
 133   simdata_task_t simdata = xbt_new0(s_simdata_task_t, 1);
 134   m_task_t task = xbt_new0(s_m_task_t, 1);
 135   task->simdata = simdata;
 136
 137   /* Task structure */
 138   task->name = xbt_strdup(name);
 139   task->data = data;
 140
 141   /* Simulator Data */
 142   simdata->computation_amount = 0;
 143   simdata->message_size = 0;
 144   simdata->cond = SIMIX_cond_init();
 145   simdata->mutex = SIMIX_mutex_init();
 146   simdata->compute = NULL;
 147   simdata->comm = NULL;
 148   simdata->rate = -1.0;
 149   simdata->refcount = 1;
 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   m_process_t self = MSG_process_self();
 169   e_surf_action_state_t state = SURF_ACTION_NOT_IN_THE_SYSTEM;
 170   CHECK_HOST();
 171
 172   simdata = task->simdata;
 173
 174   xbt_assert0((!simdata->compute)
 175               && (task->simdata->refcount == 1),
 176               "This task is executed somewhere else. Go fix your code!");
 177
 178   xbt_assert0(simdata->host_nb, "This is not a parallel task. Go to hell.");
 179
 180   DEBUG1("Computing on %s", MSG_process_self()->simdata->m_host->name);
 181
 182   simdata->refcount++;
 183
 184   SIMIX_mutex_lock(simdata->mutex);
 185   simdata->compute =
 186     SIMIX_action_parallel_execute(task->name, simdata->host_nb,
 187                                   simdata->host_list, simdata->comp_amount,
 188                                   simdata->comm_amount, 1.0, -1.0);
 189
 190   self->simdata->waiting_action = simdata->compute;
 191   SIMIX_register_action_to_condition(simdata->compute, simdata->cond);
 192   do {
 193     SIMIX_cond_wait(simdata->cond, simdata->mutex);
 194     state = SIMIX_action_get_state(task->simdata->compute);
 195   } while (state == SURF_ACTION_READY || state == SURF_ACTION_RUNNING);
 196
 197   SIMIX_unregister_action_to_condition(simdata->compute, simdata->cond);
 198   self->simdata->waiting_action = NULL;
 199
 200
 201   SIMIX_mutex_unlock(simdata->mutex);
 202   simdata->refcount--;
 203
 204   if (SIMIX_action_get_state(task->simdata->compute) == SURF_ACTION_DONE) {
 205     /* action ended, set comm and compute = NULL, the actions is already destroyed in the main function */
 206     SIMIX_action_destroy(task->simdata->compute);
 207     simdata->computation_amount = 0.0;
 208     simdata->comm = NULL;
 209     simdata->compute = NULL;
 210     MSG_RETURN(MSG_OK);
 211   } else if (SIMIX_host_get_state(SIMIX_host_self()) == 0) {
 212     /* action ended, set comm and compute = NULL, the actions is already destroyed in the main function */
 213     SIMIX_action_destroy(task->simdata->compute);
 214     simdata->comm = NULL;
 215     simdata->compute = NULL;
 216     MSG_RETURN(MSG_HOST_FAILURE);
 217   } else {
 218     /* action ended, set comm and compute = NULL, the actions is already destroyed in the main function */
 219     SIMIX_action_destroy(task->simdata->compute);
 220     simdata->comm = NULL;
 221     simdata->compute = NULL;
 222     MSG_RETURN(MSG_TASK_CANCELLED);
 223   }
 224
 225 }
 226
 227
 228 /** \ingroup msg_gos_functions
 229  * \brief Sleep for the specified number of seconds
 230  *
 231  * Makes the current process sleep until \a time seconds have elapsed.
 232  *
 233  * \param nb_sec a number of second
 234  */
 235 MSG_error_t MSG_process_sleep(double nb_sec)
 236 {
 237   smx_action_t act_sleep;
 238   m_process_t proc = MSG_process_self();
 239   e_surf_action_state_t state = SURF_ACTION_NOT_IN_THE_SYSTEM;
 240   smx_mutex_t mutex;
 241   smx_cond_t cond;
 242
 243 #ifdef HAVE_TRACING
 244   TRACE_msg_process_sleep_in (MSG_process_self());
 245 #endif
 246
 247   /* create action to sleep */
 248   act_sleep =
 249     SIMIX_action_sleep(SIMIX_process_get_host(proc->simdata->s_process),
 250                        nb_sec);
 251
 252   mutex = SIMIX_mutex_init();
 253   SIMIX_mutex_lock(mutex);
 254
 255   /* create conditional and register action to it */
 256   cond = SIMIX_cond_init();
 257
 258   proc->simdata->waiting_action = act_sleep;
 259   SIMIX_register_action_to_condition(act_sleep, cond);
 260   do {
 261     SIMIX_cond_wait(cond, mutex);
 262     state = SIMIX_action_get_state(act_sleep);
 263   } while (state == SURF_ACTION_READY || state == SURF_ACTION_RUNNING);
 264   proc->simdata->waiting_action = NULL;
 265   SIMIX_unregister_action_to_condition(act_sleep, cond);
 266   SIMIX_mutex_unlock(mutex);
 267
 268   /* remove variables */
 269   SIMIX_cond_destroy(cond);
 270   SIMIX_mutex_destroy(mutex);
 271
 272   if (SIMIX_action_get_state(act_sleep) == SURF_ACTION_DONE) {
 273     if (SIMIX_host_get_state(SIMIX_host_self()) == SURF_RESOURCE_OFF) {
 274       SIMIX_action_destroy(act_sleep);
 275 #ifdef HAVE_TRACING
 276       TRACE_msg_process_sleep_out (MSG_process_self());
 277 #endif
 278       MSG_RETURN(MSG_HOST_FAILURE);
 279     }
 280   } else {
 281     SIMIX_action_destroy(act_sleep);
 282 #ifdef HAVE_TRACING
 283     TRACE_msg_process_sleep_out (MSG_process_self());
 284 #endif
 285     MSG_RETURN(MSG_HOST_FAILURE);
 286   }
 287
 288   SIMIX_action_destroy(act_sleep);
 289 #ifdef HAVE_TRACING
 290   TRACE_msg_process_sleep_out (MSG_process_self());
 291 #endif
 292   MSG_RETURN(MSG_OK);
 293 }
 294
 295 /** \ingroup msg_gos_functions
 296  * \brief Listen on \a channel and waits for receiving a task from \a host.
 297  *
 298  * It takes three parameters.
 299  * \param task a memory location for storing a #m_task_t. It will
 300    hold a task when this function will return. Thus \a task should not
 301    be equal to \c NULL and \a *task should be equal to \c NULL. If one of
 302    those two condition does not hold, there will be a warning message.
 303  * \param channel the channel on which the agent should be
 304    listening. This value has to be >=0 and < than the maximal
 305    number of channels fixed with MSG_set_channel_number().
 306  * \param host the host that is to be watched.
 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_from_host(m_task_t * task, m_channel_t channel, m_host_t host)
 312 {
 313   return MSG_task_get_ext(task, channel, -1, host);
 314 }
 315
 316 /** \ingroup msg_gos_functions
 317  * \brief Listen on a channel and wait for receiving a task.
 318  *
 319  * It takes two parameters.
 320  * \param task a memory location for storing a #m_task_t. It will
 321    hold a task when this function will return. Thus \a task should not
 322    be equal to \c NULL and \a *task should be equal to \c NULL. If one of
 323    those two condition does not hold, there will be a warning message.
 324  * \param channel the channel on which the agent should be
 325    listening. This value has to be >=0 and < than the maximal
 326    number of channels fixed with MSG_set_channel_number().
 327  * \return #MSG_FATAL if \a task is equal to \c NULL, #MSG_WARNING
 328  * if \a *task is not equal to \c NULL, and #MSG_OK otherwise.
 329  */
 330 MSG_error_t MSG_task_get(m_task_t * task, m_channel_t channel)
 331 {
 332   return MSG_task_get_with_timeout(task, channel, -1);
 333 }
 334
 335 /** \ingroup msg_gos_functions
 336  * \brief Listen on a channel and wait for receiving a task with a timeout.
 337  *
 338  * It takes three parameters.
 339  * \param task a memory location for storing a #m_task_t. It will
 340    hold a task when this function will return. Thus \a task should not
 341    be equal to \c NULL and \a *task should be equal to \c NULL. If one of
 342    those two condition does not hold, there will be a warning message.
 343  * \param channel the channel on which the agent should be
 344    listening. This value has to be >=0 and < than the maximal
 345    number of channels fixed with MSG_set_channel_number().
 346  * \param max_duration the maximum time to wait for a task before giving
 347     up. In such a case, #MSG_TRANSFER_FAILURE will be returned, \a task
 348     will not be modified and will still be
 349     equal to \c NULL when returning.
 350  * \return #MSG_FATAL if \a task is equal to \c NULL, #MSG_WARNING
 351    if \a *task is not equal to \c NULL, and #MSG_OK otherwise.
 352  */
 353 MSG_error_t
 354 MSG_task_get_with_timeout(m_task_t * task, m_channel_t channel,
 355                           double max_duration)
 356 {
 357   return MSG_task_get_ext(task, channel, max_duration, NULL);
 358 }
 359
 360 /** \defgroup msg_gos_functions MSG Operating System Functions
 361  *  \brief This section describes the functions that can be used
 362  *  by an agent for handling some task.
 363  */
 364
 365 MSG_error_t
 366 MSG_task_get_ext(m_task_t * task, m_channel_t channel, double timeout,
 367                  m_host_t host)
 368 {
 369   xbt_assert1((channel >= 0)
 370               && (channel < msg_global->max_channel), "Invalid channel %d",
 371               channel);
 372
 373   return
 374     MSG_mailbox_get_task_ext(MSG_mailbox_get_by_channel
 375                              (MSG_host_self(), channel), task, host, timeout);
 376 }
 377
 378 MSG_error_t
 379 MSG_task_receive_from_host(m_task_t * task, const char *alias, m_host_t host)
 380 {
 381   return MSG_task_receive_ext(task, alias, -1, host);
 382 }
 383
 384 MSG_error_t MSG_task_receive(m_task_t * task, const char *alias)
 385 {
 386   return MSG_task_receive_with_timeout(task, alias, -1);
 387 }
 388
 389 MSG_error_t
 390 MSG_task_receive_with_timeout(m_task_t * task, const char *alias,
 391                               double timeout)
 392 {
 393   return MSG_task_receive_ext(task, alias, timeout, NULL);
 394 }
 395
 396 MSG_error_t
 397 MSG_task_receive_ext(m_task_t * task, const char *alias, double timeout,
 398                      m_host_t host)
 399 {
 400   return MSG_mailbox_get_task_ext(MSG_mailbox_get_by_alias(alias), task, host,
 401                                   timeout);
 402 }
 403
 404
 405 /** \ingroup msg_gos_functions
 406  * \brief Put a task on a channel of an host and waits for the end of the
 407  * transmission.
 408  *
 409  * This function is used for describing the behavior of an agent. It
 410  * takes three parameter.
 411  * \param task a #m_task_t to send on another location. This task
 412    will not be usable anymore when the function will return. There is
 413    no automatic task duplication and you have to save your parameters
 414    before calling this function. Tasks are unique and once it has been
 415    sent to another location, you should not access it anymore. You do
 416    not need to call MSG_task_destroy() but to avoid using, as an
 417    effect of inattention, this task anymore, you definitely should
 418    renitialize it with #MSG_TASK_UNINITIALIZED. Note that this task
 419    can be transfered iff it has been correctly created with
 420    MSG_task_create().
 421  * \param dest the destination of the message
 422  * \param channel the channel on which the agent should put this
 423    task. This value has to be >=0 and < than the maximal number of
 424    channels fixed with MSG_set_channel_number().
 425  * \return #MSG_FATAL if \a task is not properly initialized and
 426  * #MSG_OK otherwise. Returns #MSG_HOST_FAILURE if the host on which
 427  * this function was called was shut down. Returns
 428  * #MSG_TRANSFER_FAILURE if the transfer could not be properly done
 429  * (network failure, dest failure)
 430  */
 431 MSG_error_t MSG_task_put(m_task_t task, m_host_t dest, m_channel_t channel)
 432 {
 433   return MSG_task_put_with_timeout(task, dest, channel, -1.0);
 434 }
 435
 436 /** \ingroup msg_gos_functions
 437  * \brief Does exactly the same as MSG_task_put but with a bounded transmition
 438  * rate.
 439  *
 440  * \sa MSG_task_put
 441  */
 442 MSG_error_t
 443 MSG_task_put_bounded(m_task_t task, m_host_t dest, m_channel_t channel,
 444                      double maxrate)
 445 {
 446   task->simdata->rate = maxrate;
 447   return MSG_task_put(task, dest, channel);
 448 }
 449
 450 /** \ingroup msg_gos_functions \brief Put a task on a channel of an
 451  * host (with a timeout on the waiting of the destination host) and
 452  * waits for the end of the transmission.
 453  *
 454  * This function is used for describing the behavior of an agent. It
 455  * takes four parameter.
 456  * \param task a #m_task_t to send on another location. This task
 457    will not be usable anymore when the function will return. There is
 458    no automatic task duplication and you have to save your parameters
 459    before calling this function. Tasks are unique and once it has been
 460    sent to another location, you should not access it anymore. You do
 461    not need to call MSG_task_destroy() but to avoid using, as an
 462    effect of inattention, this task anymore, you definitely should
 463    renitialize it with #MSG_TASK_UNINITIALIZED. Note that this task
 464    can be transfered iff it has been correctly created with
 465    MSG_task_create().
 466  * \param dest the destination of the message
 467  * \param channel the channel on which the agent should put this
 468    task. This value has to be >=0 and < than the maximal number of
 469    channels fixed with MSG_set_channel_number().
 470  * \param timeout the maximum time to wait for a task before giving
 471     up. In such a case, #MSG_TRANSFER_FAILURE will be returned, \a task
 472     will not be modified
 473  * \return #MSG_FATAL if \a task is not properly initialized and
 474    #MSG_OK otherwise. Returns #MSG_HOST_FAILURE if the host on which
 475    this function was called was shut down. Returns
 476    #MSG_TRANSFER_FAILURE if the transfer could not be properly done
 477    (network failure, dest failure, timeout...)
 478  */
 479 MSG_error_t
 480 MSG_task_put_with_timeout(m_task_t task, m_host_t dest, m_channel_t channel,
 481                           double timeout)
 482 {
 483   xbt_assert1((channel >= 0)
 484               && (channel < msg_global->max_channel), "Invalid channel %d",
 485               channel);
 486
 487   return
 488     MSG_mailbox_put_with_timeout(MSG_mailbox_get_by_channel(dest, channel),
 489                                  task, timeout);
 490 }
 491
 492 MSG_error_t MSG_task_send(m_task_t task, const char *alias)
 493 {
 494   return MSG_task_send_with_timeout(task, alias, -1);
 495 }
 496
 497
 498 MSG_error_t
 499 MSG_task_send_bounded(m_task_t task, const char *alias, double maxrate)
 500 {
 501   task->simdata->rate = maxrate;
 502   return MSG_task_send(task, alias);
 503 }
 504
 505
 506 MSG_error_t
 507 MSG_task_send_with_timeout(m_task_t task, const char *alias, double timeout)
 508 {
 509   return MSG_mailbox_put_with_timeout(MSG_mailbox_get_by_alias(alias), task,
 510                                       timeout);
 511 }
 512
 513 int MSG_task_listen(const char *alias)
 514 {
 515   CHECK_HOST();
 516
 517   return !MSG_mailbox_is_empty(MSG_mailbox_get_by_alias(alias));
 518 }
 519
 520 /** \ingroup msg_gos_functions
 521  * \brief Test whether there is a pending communication on a channel.
 522  *
 523  * It takes one parameter.
 524  * \param channel the channel on which the agent should be
 525    listening. This value has to be >=0 and < than the maximal
 526    number of channels fixed with MSG_set_channel_number().
 527  * \return 1 if there is a pending communication and 0 otherwise
 528  */
 529 int MSG_task_Iprobe(m_channel_t channel)
 530 {
 531   xbt_assert1((channel >= 0)
 532               && (channel < msg_global->max_channel), "Invalid channel %d",
 533               channel);
 534
 535   CHECK_HOST();
 536
 537   return
 538     !MSG_mailbox_is_empty(MSG_mailbox_get_by_channel
 539                           (MSG_host_self(), channel));
 540 }
 541
 542 /** \ingroup msg_gos_functions
 543
 544  * \brief Return the number of tasks waiting to be received on a \a
 545    channel and sent by \a host.
 546  *
 547  * It takes two parameters.
 548  * \param channel the channel on which the agent should be
 549    listening. This value has to be >=0 and < than the maximal
 550    number of channels fixed with MSG_set_channel_number().
 551  * \param host the host that is to be watched.
 552  * \return the number of tasks waiting to be received on \a channel
 553    and sent by \a host.
 554  */
 555 int MSG_task_probe_from_host(int channel, m_host_t host)
 556 {
 557   xbt_assert1((channel >= 0)
 558               && (channel < msg_global->max_channel), "Invalid channel %d",
 559               channel);
 560
 561   CHECK_HOST();
 562
 563   return
 564     MSG_mailbox_get_count_host_waiting_tasks(MSG_mailbox_get_by_channel
 565                                              (MSG_host_self(), channel),
 566                                              host);
 567
 568 }
 569
 570 int MSG_task_listen_from_host(const char *alias, m_host_t host)
 571 {
 572   CHECK_HOST();
 573
 574   return
 575     MSG_mailbox_get_count_host_waiting_tasks(MSG_mailbox_get_by_alias(alias),
 576                                              host);
 577 }
 578
 579 /** \ingroup msg_gos_functions
 580  * \brief Test whether there is a pending communication on a channel, and who sent it.
 581  *
 582  * It takes one parameter.
 583  * \param channel the channel on which the agent should be
 584    listening. This value has to be >=0 and < than the maximal
 585    number of channels fixed with MSG_set_channel_number().
 586  * \return -1 if there is no pending communication and the PID of the process who sent it otherwise
 587  */
 588 int MSG_task_probe_from(m_channel_t channel)
 589 {
 590   m_task_t task;
 591
 592   CHECK_HOST();
 593
 594   xbt_assert1((channel >= 0)
 595               && (channel < msg_global->max_channel), "Invalid channel %d",
 596               channel);
 597
 598   if (NULL ==
 599       (task =
 600        MSG_mailbox_get_head(MSG_mailbox_get_by_channel
 601                             (MSG_host_self(), channel))))
 602     return -1;
 603
 604   return MSG_process_get_PID(task->simdata->sender);
 605 }
 606
 607 int MSG_task_listen_from(const char *alias)
 608 {
 609   m_task_t task;
 610
 611   CHECK_HOST();
 612
 613   if (NULL == (task = MSG_mailbox_get_head(MSG_mailbox_get_by_alias(alias))))
 614     return -1;
 615
 616   return MSG_process_get_PID(task->simdata->sender);
 617 }
 618
 619 /** \ingroup msg_gos_functions
 620  * \brief Wait for at most \a max_duration second for a task reception
 621    on \a channel.
 622
 623  * \a PID is updated with the PID of the first process that triggered this event if any.
 624  *
 625  * It takes three parameters:
 626  * \param channel the channel on which the agent should be
 627    listening. This value has to be >=0 and < than the maximal.
 628    number of channels fixed with MSG_set_channel_number().
 629  * \param PID a memory location for storing an int.
 630  * \param timeout the maximum time to wait for a task before
 631     giving up. In the case of a reception, *\a PID will be updated
 632     with the PID of the first process to send a task.
 633  * \return #MSG_HOST_FAILURE if the host is shut down in the meantime
 634    and #MSG_OK otherwise.
 635  */
 636 MSG_error_t
 637 MSG_channel_select_from(m_channel_t channel, double timeout, int *PID)
 638 {
 639   m_host_t h = NULL;
 640   simdata_host_t h_simdata = NULL;
 641   m_task_t t;
 642   int first_time = 1;
 643   smx_cond_t cond;
 644   msg_mailbox_t mailbox;
 645
 646   xbt_assert1((channel >= 0)
 647               && (channel < msg_global->max_channel), "Invalid channel %d",
 648               channel);
 649
 650   if (PID) {
 651     *PID = -1;
 652   }
 653
 654   if (timeout == 0.0) {
 655     *PID = MSG_task_probe_from(channel);
 656     MSG_RETURN(MSG_OK);
 657   } else {
 658     CHECK_HOST();
 659     h = MSG_host_self();
 660     h_simdata = h->simdata;
 661
 662     mailbox = MSG_mailbox_get_by_channel(MSG_host_self(), channel);
 663
 664     while (MSG_mailbox_is_empty(mailbox)) {
 665       if (timeout > 0) {
 666         if (!first_time) {
 667           MSG_RETURN(MSG_OK);
 668         }
 669       }
 670
 671       SIMIX_mutex_lock(h_simdata->mutex);
 672
 673       xbt_assert1(!MSG_mailbox_get_cond(mailbox),
 674                   "A process is already blocked on this channel %d", channel);
 675
 676       cond = SIMIX_cond_init();
 677
 678       MSG_mailbox_set_cond(mailbox, cond);
 679
 680       if (timeout > 0) {
 681         SIMIX_cond_wait_timeout(cond, h_simdata->mutex, timeout);
 682       } else {
 683         SIMIX_cond_wait(cond, h_simdata->mutex);
 684       }
 685
 686       SIMIX_cond_destroy(cond);
 687       SIMIX_mutex_unlock(h_simdata->mutex);
 688
 689       if (SIMIX_host_get_state(h_simdata->smx_host) == 0) {
 690         MSG_RETURN(MSG_HOST_FAILURE);
 691       }
 692
 693       MSG_mailbox_set_cond(mailbox, NULL);
 694       first_time = 0;
 695     }
 696
 697     if (NULL == (t = MSG_mailbox_get_head(mailbox)))
 698       MSG_RETURN(MSG_OK);
 699
 700
 701     if (PID) {
 702       *PID = MSG_process_get_PID(t->simdata->sender);
 703     }
 704
 705     MSG_RETURN(MSG_OK);
 706   }
 707 }
 708
 709
 710 MSG_error_t MSG_alias_select_from(const char *alias, double timeout, int *PID)
 711 {
 712   m_host_t h = NULL;
 713   simdata_host_t h_simdata = NULL;
 714   m_task_t t;
 715   int first_time = 1;
 716   smx_cond_t cond;
 717   msg_mailbox_t mailbox;
 718
 719   if (PID) {
 720     *PID = -1;
 721   }
 722
 723   if (timeout == 0.0) {
 724     *PID = MSG_task_listen_from(alias);
 725     MSG_RETURN(MSG_OK);
 726   } else {
 727     CHECK_HOST();
 728     h = MSG_host_self();
 729     h_simdata = h->simdata;
 730
 731     DEBUG2("Probing on alias %s (%s)", alias, h->name);
 732
 733     mailbox = MSG_mailbox_get_by_alias(alias);
 734
 735     while (MSG_mailbox_is_empty(mailbox)) {
 736       if (timeout > 0) {
 737         if (!first_time) {
 738           MSG_RETURN(MSG_OK);
 739         }
 740       }
 741
 742       SIMIX_mutex_lock(h_simdata->mutex);
 743
 744       xbt_assert1(!MSG_mailbox_get_cond(mailbox),
 745                   "A process is already blocked on this alias %s", alias);
 746
 747       cond = SIMIX_cond_init();
 748
 749       MSG_mailbox_set_cond(mailbox, cond);
 750
 751       if (timeout > 0) {
 752         SIMIX_cond_wait_timeout(cond, h_simdata->mutex, timeout);
 753       } else {
 754         SIMIX_cond_wait(cond, h_simdata->mutex);
 755       }
 756
 757       SIMIX_cond_destroy(cond);
 758       SIMIX_mutex_unlock(h_simdata->mutex);
 759
 760       if (SIMIX_host_get_state(h_simdata->smx_host) == 0) {
 761         MSG_RETURN(MSG_HOST_FAILURE);
 762       }
 763
 764       MSG_mailbox_set_cond(mailbox, NULL);
 765       first_time = 0;
 766     }
 767
 768     if (NULL == (t = MSG_mailbox_get_head(mailbox)))
 769       MSG_RETURN(MSG_OK);
 770
 771
 772     if (PID) {
 773       *PID = MSG_process_get_PID(t->simdata->sender);
 774     }
 775
 776     MSG_RETURN(MSG_OK);
 777   }
 778 }