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