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         MSG_RETURN(MSG_OK);
 256 }
 257
 258 /** \ingroup msg_gos_functions
 259  * \brief Return the number of MSG tasks currently running on
 260  * the host of the current running process.
 261  */
 262 static int
 263 MSG_get_msgload(void)
 264 {
 265         xbt_die("not implemented yet");
 266         return 0;
 267 }
 268
 269
 270
 271 /** \ingroup msg_gos_functions
 272  * \brief Listen on \a channel and waits for receiving a task from \a host.
 273  *
 274  * It takes three parameters.
 275  * \param task a memory location for storing a #m_task_t. It will
 276    hold a task when this function will return. Thus \a task should not
 277    be equal to \c NULL and \a *task should be equal to \c NULL. If one of
 278    those two condition does not hold, there will be a warning message.
 279  * \param channel the channel on which the agent should be
 280    listening. This value has to be >=0 and < than the maximal
 281    number of channels fixed with MSG_set_channel_number().
 282  * \param host the host that is to be watched.
 283  * \return #MSG_FATAL if \a task is equal to \c NULL, #MSG_WARNING
 284    if \a *task is not equal to \c NULL, and #MSG_OK otherwise.
 285  */
 286 MSG_error_t
 287 MSG_task_get_from_host(m_task_t * task, m_channel_t channel, m_host_t host)
 288 {
 289         return MSG_task_get_ext(task, channel, -1, host);
 290 }
 291
 292 /** \ingroup msg_gos_functions
 293  * \brief Listen on a channel and wait for receiving a task.
 294  *
 295  * It takes two 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  * \return #MSG_FATAL if \a task is equal to \c NULL, #MSG_WARNING
 304  * if \a *task is not equal to \c NULL, and #MSG_OK otherwise.
 305  */
 306 MSG_error_t
 307 MSG_task_get(m_task_t * task, m_channel_t channel)
 308 {
 309         return MSG_task_get_with_timeout(task, channel, -1);
 310 }
 311
 312 /** \ingroup msg_gos_functions
 313  * \brief Listen on a channel and wait for receiving a task with a timeout.
 314  *
 315  * It takes three parameters.
 316  * \param task a memory location for storing a #m_task_t. It will
 317    hold a task when this function will return. Thus \a task should not
 318    be equal to \c NULL and \a *task should be equal to \c NULL. If one of
 319    those two condition does not hold, there will be a warning message.
 320  * \param channel the channel on which the agent should be
 321    listening. This value has to be >=0 and < than the maximal
 322    number of channels fixed with MSG_set_channel_number().
 323  * \param max_duration the maximum time to wait for a task before giving
 324     up. In such a case, #MSG_TRANSFER_FAILURE will be returned, \a task
 325     will not be modified and will still be
 326     equal to \c NULL when returning.
 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
 331 MSG_task_get_with_timeout(m_task_t * task, m_channel_t channel, double max_duration)
 332 {
 333   return MSG_task_get_ext(task, channel, max_duration, NULL);
 334 }
 335
 336 /** \defgroup msg_gos_functions MSG Operating System Functions
 337  *  \brief This section describes the functions that can be used
 338  *  by an agent for handling some task.
 339  */
 340
 341 MSG_error_t
 342 MSG_task_get_ext(m_task_t * task, m_channel_t channel, double timeout,m_host_t host)
 343 {
 344         xbt_assert1((channel >= 0) && (channel < msg_global->max_channel), "Invalid channel %d",channel);
 345
 346         return MSG_mailbox_get_task_ext(MSG_mailbox_get_by_channel(MSG_host_self(), channel), task, host, timeout);
 347 }
 348
 349 MSG_error_t
 350 MSG_task_receive_from_host(m_task_t * task, const char* alias, m_host_t host)
 351 {
 352         return MSG_task_receive_ext(task, alias, -1, host);
 353 }
 354
 355 MSG_error_t
 356 MSG_task_receive(m_task_t * task, const char* alias)
 357 {
 358         return MSG_task_receive_with_timeout(task, alias, -1);
 359 }
 360
 361 MSG_error_t
 362 MSG_task_receive_with_timeout(m_task_t * task, const char* alias, double timeout)
 363 {
 364         return MSG_task_receive_ext(task, alias, timeout, NULL);
 365 }
 366
 367 MSG_error_t
 368 MSG_task_receive_ext(m_task_t* task, const char* alias, double timeout, m_host_t host)
 369 {
 370         return MSG_mailbox_get_task_ext(MSG_mailbox_get_by_alias(alias), task, host, timeout);
 371 }
 372
 373
 374 /** \ingroup msg_gos_functions
 375  * \brief Put a task on a channel of an host and waits for the end of the
 376  * transmission.
 377  *
 378  * This function is used for describing the behavior of an agent. It
 379  * takes three parameter.
 380  * \param task a #m_task_t to send on another location. This task
 381    will not be usable anymore when the function will return. There is
 382    no automatic task duplication and you have to save your parameters
 383    before calling this function. Tasks are unique and once it has been
 384    sent to another location, you should not access it anymore. You do
 385    not need to call MSG_task_destroy() but to avoid using, as an
 386    effect of inattention, this task anymore, you definitely should
 387    renitialize it with #MSG_TASK_UNINITIALIZED. Note that this task
 388    can be transfered iff it has been correctly created with
 389    MSG_task_create().
 390  * \param dest the destination of the message
 391  * \param channel the channel on which the agent should put this
 392    task. This value has to be >=0 and < than the maximal number of
 393    channels fixed with MSG_set_channel_number().
 394  * \return #MSG_FATAL if \a task is not properly initialized and
 395  * #MSG_OK otherwise. Returns #MSG_HOST_FAILURE if the host on which
 396  * this function was called was shut down. Returns
 397  * #MSG_TRANSFER_FAILURE if the transfer could not be properly done
 398  * (network failure, dest failure)
 399  */
 400 MSG_error_t
 401 MSG_task_put(m_task_t task, m_host_t dest, m_channel_t channel)
 402 {
 403         return MSG_task_put_with_timeout(task, dest, channel, -1.0);
 404 }
 405
 406 /** \ingroup msg_gos_functions
 407  * \brief Does exactly the same as MSG_task_put but with a bounded transmition
 408  * rate.
 409  *
 410  * \sa MSG_task_put
 411  */
 412 MSG_error_t
 413 MSG_task_put_bounded(m_task_t task, m_host_t dest, m_channel_t channel, double maxrate)
 414 {
 415         task->simdata->rate = maxrate;
 416         return MSG_task_put(task, dest, channel);
 417 }
 418
 419 /** \ingroup msg_gos_functions \brief Put a task on a channel of an
 420  * host (with a timeout on the waiting of the destination host) and
 421  * waits for the end of the transmission.
 422  *
 423  * This function is used for describing the behavior of an agent. It
 424  * takes four parameter.
 425  * \param task a #m_task_t to send on another location. This task
 426    will not be usable anymore when the function will return. There is
 427    no automatic task duplication and you have to save your parameters
 428    before calling this function. Tasks are unique and once it has been
 429    sent to another location, you should not access it anymore. You do
 430    not need to call MSG_task_destroy() but to avoid using, as an
 431    effect of inattention, this task anymore, you definitely should
 432    renitialize it with #MSG_TASK_UNINITIALIZED. Note that this task
 433    can be transfered iff it has been correctly created with
 434    MSG_task_create().
 435  * \param dest the destination of the message
 436  * \param channel the channel on which the agent should put this
 437    task. This value has to be >=0 and < than the maximal number of
 438    channels fixed with MSG_set_channel_number().
 439  * \param timeout the maximum time to wait for a task before giving
 440     up. In such a case, #MSG_TRANSFER_FAILURE will be returned, \a task
 441     will not be modified
 442  * \return #MSG_FATAL if \a task is not properly initialized and
 443    #MSG_OK otherwise. Returns #MSG_HOST_FAILURE if the host on which
 444    this function was called was shut down. Returns
 445    #MSG_TRANSFER_FAILURE if the transfer could not be properly done
 446    (network failure, dest failure, timeout...)
 447  */
 448 MSG_error_t
 449 MSG_task_put_with_timeout(m_task_t task, m_host_t dest, m_channel_t channel, double timeout)
 450 {
 451         xbt_assert1((channel >= 0) && (channel < msg_global->max_channel), "Invalid channel %d", channel);
 452
 453         return MSG_mailbox_put_with_timeout(MSG_mailbox_get_by_channel(dest, channel), task, timeout);
 454 }
 455
 456 MSG_error_t
 457 MSG_task_send(m_task_t task,const char* alias)
 458 {
 459         return MSG_task_send_with_timeout(task, alias, -1);
 460 }
 461
 462
 463 MSG_error_t
 464 MSG_task_send_bounded(m_task_t task, const char* alias, double maxrate)
 465 {
 466         task->simdata->rate = maxrate;
 467         return MSG_task_send(task, alias);
 468 }
 469
 470
 471 MSG_error_t
 472 MSG_task_send_with_timeout(m_task_t task, const char* alias, double timeout)
 473 {
 474         return MSG_mailbox_put_with_timeout(MSG_mailbox_get_by_alias(alias), task, timeout);
 475 }
 476
 477 int
 478 MSG_task_listen(const char* alias)
 479 {
 480         CHECK_HOST();
 481
 482         return !MSG_mailbox_is_empty(MSG_mailbox_get_by_alias(alias));
 483 }
 484
 485 /** \ingroup msg_gos_functions
 486  * \brief Test whether there is a pending communication on a channel.
 487  *
 488  * It takes one parameter.
 489  * \param channel the channel on which the agent should be
 490    listening. This value has to be >=0 and < than the maximal
 491    number of channels fixed with MSG_set_channel_number().
 492  * \return 1 if there is a pending communication and 0 otherwise
 493  */
 494 int
 495 MSG_task_Iprobe(m_channel_t channel)
 496 {
 497   xbt_assert1((channel >= 0) && (channel < msg_global->max_channel), "Invalid channel %d", channel);
 498
 499   CHECK_HOST();
 500
 501   return !MSG_mailbox_is_empty(MSG_mailbox_get_by_channel(MSG_host_self(), channel));
 502 }
 503
 504 /** \ingroup msg_gos_functions
 505
 506  * \brief Return the number of tasks waiting to be received on a \a
 507    channel and sent by \a host.
 508  *
 509  * It takes two parameters.
 510  * \param channel the channel on which the agent should be
 511    listening. This value has to be >=0 and < than the maximal
 512    number of channels fixed with MSG_set_channel_number().
 513  * \param host the host that is to be watched.
 514  * \return the number of tasks waiting to be received on \a channel
 515    and sent by \a host.
 516  */
 517 int
 518 MSG_task_probe_from_host(int channel, m_host_t host)
 519 {
 520         xbt_assert1((channel >= 0) && (channel < msg_global->max_channel), "Invalid channel %d", channel);
 521
 522         CHECK_HOST();
 523
 524         return MSG_mailbox_get_count_host_waiting_tasks(MSG_mailbox_get_by_channel(MSG_host_self(), channel),host);
 525
 526 }
 527
 528 int
 529 MSG_task_listen_from_host(const char* alias, m_host_t host)
 530 {
 531         CHECK_HOST();
 532
 533         return MSG_mailbox_get_count_host_waiting_tasks(MSG_mailbox_get_by_alias(alias),host);
 534 }
 535
 536 /** \ingroup msg_gos_functions
 537  * \brief Test whether there is a pending communication on a channel, and who sent it.
 538  *
 539  * It takes one parameter.
 540  * \param channel the channel on which the agent should be
 541    listening. This value has to be >=0 and < than the maximal
 542    number of channels fixed with MSG_set_channel_number().
 543  * \return -1 if there is no pending communication and the PID of the process who sent it otherwise
 544  */
 545 int
 546 MSG_task_probe_from(m_channel_t channel)
 547 {
 548         m_task_t task;
 549
 550         CHECK_HOST();
 551
 552         xbt_assert1((channel >= 0) && (channel < msg_global->max_channel), "Invalid channel %d", channel);
 553
 554         if(NULL == (task = MSG_mailbox_get_head(MSG_mailbox_get_by_channel(MSG_host_self(), channel))))
 555                 return -1;
 556
 557         return MSG_process_get_PID(task->simdata->sender);
 558 }
 559
 560 int
 561 MSG_task_listen_from(const char* alias)
 562 {
 563         m_task_t task;
 564
 565         CHECK_HOST();
 566
 567         if(NULL == (task = MSG_mailbox_get_head(MSG_mailbox_get_by_alias(alias))))
 568                 return -1;
 569
 570         return MSG_process_get_PID(task->simdata->sender);
 571 }
 572
 573 /** \ingroup msg_gos_functions
 574  * \brief Wait for at most \a max_duration second for a task reception
 575    on \a channel.
 576
 577  * \a PID is updated with the PID of the first process that triggered this event if any.
 578  *
 579  * It takes three parameters:
 580  * \param channel the channel on which the agent should be
 581    listening. This value has to be >=0 and < than the maximal.
 582    number of channels fixed with MSG_set_channel_number().
 583  * \param PID a memory location for storing an int.
 584  * \param timeout the maximum time to wait for a task before
 585     giving up. In the case of a reception, *\a PID will be updated
 586     with the PID of the first process to send a task.
 587  * \return #MSG_HOST_FAILURE if the host is shut down in the meantime
 588    and #MSG_OK otherwise.
 589  */
 590 MSG_error_t
 591 MSG_channel_select_from(m_channel_t channel,double timeout, int *PID)
 592 {
 593         m_host_t h = NULL;
 594         simdata_host_t h_simdata = NULL;
 595         m_task_t t;
 596         int first_time = 1;
 597         smx_cond_t cond;
 598         msg_mailbox_t mailbox;
 599
 600         xbt_assert1((channel >= 0) && (channel < msg_global->max_channel), "Invalid channel %d",channel);
 601
 602         if(PID)
 603         {
 604                 *PID = -1;
 605         }
 606
 607         if (timeout == 0.0)
 608         {
 609                 *PID = MSG_task_probe_from(channel);
 610                 MSG_RETURN(MSG_OK);
 611         }
 612         else
 613         {
 614                 CHECK_HOST();
 615                 h = MSG_host_self();
 616                 h_simdata = h->simdata;
 617
 618                 mailbox = MSG_mailbox_get_by_channel(MSG_host_self(), channel);
 619
 620                 while(MSG_mailbox_is_empty(mailbox))
 621                 {
 622                         if(timeout > 0)
 623                         {
 624                                 if (!first_time)
 625                                 {
 626                                         MSG_RETURN(MSG_OK);
 627                                 }
 628                         }
 629
 630                         SIMIX_mutex_lock(h_simdata->mutex);
 631
 632                         xbt_assert1(!MSG_mailbox_get_cond(mailbox),"A process is already blocked on this channel %d",channel);
 633
 634                         cond = SIMIX_cond_init();
 635
 636                         MSG_mailbox_set_cond(mailbox, cond);
 637
 638                         if (timeout > 0)
 639                         {
 640                                 SIMIX_cond_wait_timeout(cond, h_simdata->mutex, timeout);
 641                         }
 642                         else
 643                         {
 644                                 SIMIX_cond_wait(cond, h_simdata->mutex);
 645                         }
 646
 647                         SIMIX_cond_destroy(cond);
 648                         SIMIX_mutex_unlock(h_simdata->mutex);
 649
 650                         if (SIMIX_host_get_state(h_simdata->smx_host) == 0)
 651                         {
 652                                 MSG_RETURN(MSG_HOST_FAILURE);
 653                         }
 654
 655                         MSG_mailbox_set_cond(mailbox,NULL);
 656                         first_time = 0;
 657                 }
 658
 659                 if(NULL == (t = MSG_mailbox_get_head(mailbox)))
 660                         MSG_RETURN(MSG_OK);
 661
 662
 663                 if (PID)
 664                 {
 665                         *PID = MSG_process_get_PID(t->simdata->sender);
 666                 }
 667
 668                 MSG_RETURN(MSG_OK);
 669         }
 670 }
 671
 672
 673 MSG_error_t
 674 MSG_alias_select_from(const char* alias, double timeout, int* PID)
 675 {
 676         m_host_t h = NULL;
 677         simdata_host_t h_simdata = NULL;
 678         m_task_t t;
 679         int first_time = 1;
 680         smx_cond_t cond;
 681         msg_mailbox_t mailbox;
 682
 683         if (PID)
 684         {
 685                 *PID = -1;
 686         }
 687
 688         if(timeout == 0.0)
 689         {
 690                 *PID = MSG_task_listen_from(alias);
 691                 MSG_RETURN(MSG_OK);
 692         }
 693         else
 694         {
 695                 CHECK_HOST();
 696                 h = MSG_host_self();
 697                 h_simdata = h->simdata;
 698
 699                 DEBUG2("Probing on alias %s (%s)", alias, h->name);
 700
 701                 mailbox = MSG_mailbox_get_by_alias(alias);
 702
 703                 while(MSG_mailbox_is_empty(mailbox))
 704                 {
 705                         if(timeout > 0)
 706                         {
 707                                 if (!first_time)
 708                                 {
 709                                         MSG_RETURN(MSG_OK);
 710                                 }
 711                         }
 712
 713                         SIMIX_mutex_lock(h_simdata->mutex);
 714
 715                         xbt_assert1(!MSG_mailbox_get_cond(mailbox),"A process is already blocked on this alias %s",alias);
 716
 717                         cond = SIMIX_cond_init();
 718
 719                         MSG_mailbox_set_cond(mailbox, cond);
 720
 721                         if (timeout > 0)
 722                         {
 723                                 SIMIX_cond_wait_timeout(cond, h_simdata->mutex, timeout);
 724                         }
 725                         else
 726                         {
 727                                 SIMIX_cond_wait(cond, h_simdata->mutex);
 728                         }
 729
 730                         SIMIX_cond_destroy(cond);
 731                         SIMIX_mutex_unlock(h_simdata->mutex);
 732
 733                         if (SIMIX_host_get_state(h_simdata->smx_host) == 0)
 734                         {
 735                                 MSG_RETURN(MSG_HOST_FAILURE);
 736                         }
 737
 738                         MSG_mailbox_set_cond(mailbox,NULL);
 739                         first_time = 0;
 740                 }
 741
 742                 if(NULL == (t = MSG_mailbox_get_head(mailbox)))
 743                         MSG_RETURN(MSG_OK);
 744
 745
 746                 if (PID)
 747                 {
 748                         *PID = MSG_process_get_PID(t->simdata->sender);
 749                 }
 750
 751                 MSG_RETURN(MSG_OK);
 752         }
 753 }
 754
 755
 756
 757
 758