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