Logo AND Algorithmique Numérique Distribuée

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