Logo AND Algorithmique Numérique Distribuée

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