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         DEBUG1("MSG_task_receive_ext: Trying to receive a message on mailbox '%s'",alias);
 409         return MSG_mailbox_get_task_ext(MSG_mailbox_get_by_alias(alias), task, host,
 410                         timeout);
 411 }
 412
 413 /** \ingroup msg_gos_functions
 414  * \brief Send a task on a channel.
 415  *
 416  * This function takes two parameter.
 417  * \param task a #m_task_t to send on another location.
 418  * \param alias the channel on which the agent should put this
 419  task. This value has to be >=0 and < than the maximal number of
 420  channels fixed with MSG_set_channel_number().
 421  * \return the msg_comm_t communication.
 422  */
 423 msg_comm_t MSG_task_isend(m_task_t task, const char *alias) {
 424         simdata_task_t t_simdata = NULL;
 425         m_process_t process = MSG_process_self();
 426         msg_mailbox_t mailbox = MSG_mailbox_get_by_alias(alias);
 427
 428         CHECK_HOST();
 429
 430         /* FIXME: these functions are not tracable */
 431
 432         /* Prepare the task to send */
 433         t_simdata = task->simdata;
 434         t_simdata->sender = process;
 435         t_simdata->source = MSG_host_self();
 436
 437         xbt_assert0(t_simdata->refcount == 1,
 438                         "This task is still being used somewhere else. You cannot send it now. Go fix your code!");
 439
 440         t_simdata->refcount++;
 441         msg_global->sent_msg++;
 442         process->simdata->waiting_task = task;
 443
 444         /* Send it by calling SIMIX network layer */
 445
 446         /* Kept for semantical compatibility with older implementation */
 447         if(mailbox->cond)
 448                 SIMIX_cond_signal(mailbox->cond);
 449
 450         return SIMIX_network_isend(mailbox->rdv, t_simdata->message_size, t_simdata->rate,
 451                         task, sizeof(void*), &t_simdata->comm);
 452 }
 453
 454 /** \ingroup msg_gos_functions
 455  * \brief Listen on a channel for receiving a task from an asynchronous communication.
 456  *
 457  * It takes two parameters.
 458  * \param task a memory location for storing a #m_task_t.
 459  * \param alias the channel on which the agent should be
 460  listening. This value has to be >=0 and < than the maximal
 461  number of channels fixed with MSG_set_channel_number().
 462  * \return the msg_comm_t communication.
 463  */
 464 msg_comm_t MSG_task_irecv(m_task_t * task, const char *alias) {
 465         smx_comm_t comm;
 466         smx_rdv_t rdv = MSG_mailbox_get_by_alias(alias)->rdv;
 467         msg_mailbox_t mailbox=MSG_mailbox_get_by_alias(alias);
 468         size_t size = sizeof(void*);
 469
 470         CHECK_HOST();
 471
 472         /* FIXME: these functions are not tracable */
 473
 474         memset(&comm,0,sizeof(comm));
 475
 476         /* Kept for compatibility with older implementation */
 477         xbt_assert1(!MSG_mailbox_get_cond(mailbox),
 478                         "A process is already blocked on this channel %s",
 479                         MSG_mailbox_get_alias(mailbox));
 480
 481         /* Sanity check */
 482         xbt_assert0(task, "Null pointer for the task storage");
 483
 484         if (*task)
 485                 CRITICAL0("MSG_task_get() was asked to write in a non empty task struct.");
 486
 487         /* Try to receive it by calling SIMIX network layer */
 488         return SIMIX_network_irecv(rdv, task, &size);
 489 }
 490 /** \ingroup msg_gos_functions
 491  * \brief Test the status of a communication.
 492  *
 493  * It takes one parameter.
 494  * \param comm the communication to test.
 495  * \return the status of the communication:
 496  *              TRUE : the communication is completed
 497  *              FALSE: the communication is incompleted
 498  * If the status is FALSE, don't forget to use MSG_process_sleep() after the test.
 499  */
 500 int MSG_comm_test(msg_comm_t comm) {
 501         return SIMIX_network_test(comm);
 502 }
 503 /** \ingroup msg_gos_functions
 504  * \brief After received TRUE to MSG_comm_test(), the communication must be destroyed.
 505  *
 506  * It takes one parameter.
 507  * \param comm the communication to destroy.
 508  */
 509 void MSG_comm_destroy(msg_comm_t comm) {
 510         if(!(comm->src_proc == SIMIX_process_self()))
 511         {
 512                 m_task_t  task;
 513                 task = (m_task_t) SIMIX_communication_get_src_buf(comm);
 514                 task->simdata->refcount--;
 515         }
 516         SIMIX_communication_destroy(comm);
 517 }
 518
 519 /** \ingroup msg_gos_functions
 520  * \brief Wait for the completion of a communication.
 521  *
 522  * It takes two parameters.
 523  * \param comm the communication to wait.
 524  * \param timeout Wait until the communication terminates or the timeout occurs
 525  * \return MSG_error_t
 526  */
 527 MSG_error_t MSG_comm_wait(msg_comm_t comm, double timeout) {
 528         xbt_ex_t e;
 529         MSG_error_t res = MSG_OK;
 530         TRY {
 531                 SIMIX_network_wait(comm,timeout);
 532
 533                 if(!(comm->src_proc == SIMIX_process_self()))
 534                 {
 535                         m_task_t  task;
 536                         task = (m_task_t) SIMIX_communication_get_src_buf(comm);
 537                         task->simdata->refcount--;
 538                 }
 539
 540                 /* FIXME: these functions are not tracable */
 541         }  CATCH(e){
 542                 switch(e.category){
 543                         case host_error:
 544                                 res = MSG_HOST_FAILURE;
 545                                 break;
 546                         case network_error:
 547                                 res = MSG_TRANSFER_FAILURE;
 548                                 break;
 549                         case timeout_error:
 550                                 res = MSG_TIMEOUT;
 551                                 break;
 552                         default:
 553                                 xbt_die(bprintf("Unhandled SIMIX network exception: %s",e.msg));
 554                 }
 555                 xbt_ex_free(e);
 556         }
 557         return res;
 558 }
 559
 560 /** \ingroup msg_gos_functions
 561 * \brief This function is called by a sender and permit to wait for each communication
 562 *
 563 * It takes three parameters.
 564 * \param comm a vector of communication
 565 * \param nb_elem is the size of the comm vector
 566 * \param timeout for each call of  MSG_comm_wait
 567 */
 568 void MSG_comm_waitall(msg_comm_t *comm,int nb_elem, double timeout) {
 569         int i=0;
 570         for(i=0; i<nb_elem; i++)
 571         {
 572                 MSG_comm_wait(comm[i],timeout);
 573         }
 574 }
 575 /** \ingroup msg_gos_functions
 576 * \brief This function wait for the first completed communication
 577 *
 578 * It takes on parameter.
 579 * \param comms a vector of communication
 580 * \return the position of the completed communication from the xbt_dynar_t.
 581 */
 582 int MSG_comm_waitany(xbt_dynar_t comms) {
 583         return SIMIX_network_waitany(comms);
 584 }
 585
 586 m_task_t MSG_comm_get_task(msg_comm_t comm) {
 587         xbt_assert0(comm, "Invalid parameters");
 588         return (m_task_t) SIMIX_communication_get_src_buf(comm);
 589 }
 590
 591 /** \ingroup msg_gos_functions
 592  * \brief Put a task on a channel of an host and waits for the end of the
 593  * transmission.
 594  *
 595  * This function is used for describing the behavior of an agent. It
 596  * takes three parameter.
 597  * \param task a #m_task_t to send on another location. This task
 598  will not be usable anymore when the function will return. There is
 599  no automatic task duplication and you have to save your parameters
 600  before calling this function. Tasks are unique and once it has been
 601  sent to another location, you should not access it anymore. You do
 602  not need to call MSG_task_destroy() but to avoid using, as an
 603  effect of inattention, this task anymore, you definitely should
 604  renitialize it with #MSG_TASK_UNINITIALIZED. Note that this task
 605  can be transfered iff it has been correctly created with
 606  MSG_task_create().
 607  * \param dest the destination of the message
 608  * \param channel the channel on which the agent should put this
 609  task. This value has to be >=0 and < than the maximal number of
 610  channels fixed with MSG_set_channel_number().
 611  * \return #MSG_FATAL if \a task is not properly initialized and
 612  * #MSG_OK otherwise. Returns #MSG_HOST_FAILURE if the host on which
 613  * this function was called was shut down. Returns
 614  * #MSG_TRANSFER_FAILURE if the transfer could not be properly done
 615  * (network failure, dest failure)
 616  */
 617 MSG_error_t MSG_task_put(m_task_t task, m_host_t dest, m_channel_t channel)
 618 {
 619         return MSG_task_put_with_timeout(task, dest, channel, -1.0);
 620 }
 621
 622 /** \ingroup msg_gos_functions
 623  * \brief Does exactly the same as MSG_task_put but with a bounded transmition
 624  * rate.
 625  *
 626  * \sa MSG_task_put
 627  */
 628         MSG_error_t
 629 MSG_task_put_bounded(m_task_t task, m_host_t dest, m_channel_t channel,
 630                 double maxrate)
 631 {
 632         task->simdata->rate = maxrate;
 633         return MSG_task_put(task, dest, channel);
 634 }
 635
 636 /** \ingroup msg_gos_functions \brief Put a task on a channel of an
 637  * host (with a timeout on the waiting of the destination host) and
 638  * waits for the end of the transmission.
 639  *
 640  * This function is used for describing the behavior of an agent. It
 641  * takes four parameter.
 642  * \param task a #m_task_t to send on another location. This task
 643  will not be usable anymore when the function will return. There is
 644  no automatic task duplication and you have to save your parameters
 645  before calling this function. Tasks are unique and once it has been
 646  sent to another location, you should not access it anymore. You do
 647  not need to call MSG_task_destroy() but to avoid using, as an
 648  effect of inattention, this task anymore, you definitely should
 649  renitialize it with #MSG_TASK_UNINITIALIZED. Note that this task
 650  can be transfered iff it has been correctly created with
 651  MSG_task_create().
 652  * \param dest the destination of the message
 653  * \param channel the channel on which the agent should put this
 654  task. This value has to be >=0 and < than the maximal number of
 655  channels fixed with MSG_set_channel_number().
 656  * \param timeout the maximum time to wait for a task before giving
 657  up. In such a case, #MSG_TRANSFER_FAILURE will be returned, \a task
 658  will not be modified
 659  * \return #MSG_FATAL if \a task is not properly initialized and
 660 #MSG_OK otherwise. Returns #MSG_HOST_FAILURE if the host on which
 661 this function was called was shut down. Returns
 662 #MSG_TRANSFER_FAILURE if the transfer could not be properly done
 663 (network failure, dest failure, timeout...)
 664  */
 665         MSG_error_t
 666 MSG_task_put_with_timeout(m_task_t task, m_host_t dest, m_channel_t channel,
 667                 double timeout)
 668 {
 669         xbt_assert1((channel >= 0)
 670                         && (channel < msg_global->max_channel), "Invalid channel %d",
 671                         channel);
 672
 673         return
 674                 MSG_mailbox_put_with_timeout(MSG_mailbox_get_by_channel(dest, channel),
 675                                 task, timeout);
 676 }
 677
 678 MSG_error_t MSG_task_send(m_task_t task, const char *alias)
 679 {
 680         DEBUG1("MSG_task_send: Trying to send a message on mailbox '%s'",alias);
 681         return MSG_task_send_with_timeout(task, alias, -1);
 682 }
 683
 684
 685         MSG_error_t
 686 MSG_task_send_bounded(m_task_t task, const char *alias, double maxrate)
 687 {
 688         task->simdata->rate = maxrate;
 689         return MSG_task_send(task, alias);
 690 }
 691
 692
 693         MSG_error_t
 694 MSG_task_send_with_timeout(m_task_t task, const char *alias, double timeout)
 695 {
 696         return MSG_mailbox_put_with_timeout(MSG_mailbox_get_by_alias(alias), task,
 697                         timeout);
 698 }
 699
 700 int MSG_task_listen(const char *alias)
 701 {
 702         CHECK_HOST();
 703
 704         return !MSG_mailbox_is_empty(MSG_mailbox_get_by_alias(alias));
 705 }
 706
 707 /** \ingroup msg_gos_functions
 708  * \brief Test whether there is a pending communication on a channel.
 709  *
 710  * It takes one parameter.
 711  * \param channel the channel on which the agent should be
 712  listening. This value has to be >=0 and < than the maximal
 713  number of channels fixed with MSG_set_channel_number().
 714  * \return 1 if there is a pending communication and 0 otherwise
 715  */
 716 int MSG_task_Iprobe(m_channel_t channel)
 717 {
 718         xbt_assert1((channel >= 0)
 719                         && (channel < msg_global->max_channel), "Invalid channel %d",
 720                         channel);
 721
 722         CHECK_HOST();
 723
 724         return
 725                 !MSG_mailbox_is_empty(MSG_mailbox_get_by_channel
 726                                 (MSG_host_self(), channel));
 727 }
 728
 729 /** \ingroup msg_gos_functions
 730
 731  * \brief Return the number of tasks waiting to be received on a \a
 732  channel and sent by \a host.
 733  *
 734  * It takes two parameters.
 735  * \param channel the channel on which the agent should be
 736  listening. This value has to be >=0 and < than the maximal
 737  number of channels fixed with MSG_set_channel_number().
 738  * \param host the host that is to be watched.
 739  * \return the number of tasks waiting to be received on \a channel
 740  and sent by \a host.
 741  */
 742 int MSG_task_probe_from_host(int channel, m_host_t host)
 743 {
 744         xbt_assert1((channel >= 0)
 745                         && (channel < msg_global->max_channel), "Invalid channel %d",
 746                         channel);
 747
 748         CHECK_HOST();
 749
 750         return
 751                 MSG_mailbox_get_count_host_waiting_tasks(MSG_mailbox_get_by_channel
 752                                 (MSG_host_self(), channel),
 753                                 host);
 754
 755 }
 756
 757 int MSG_task_listen_from_host(const char *alias, m_host_t host)
 758 {
 759         CHECK_HOST();
 760
 761         return
 762                 MSG_mailbox_get_count_host_waiting_tasks(MSG_mailbox_get_by_alias(alias),
 763                                 host);
 764 }
 765
 766 /** \ingroup msg_gos_functions
 767  * \brief Test whether there is a pending communication on a channel, and who sent it.
 768  *
 769  * It takes one parameter.
 770  * \param channel the channel on which the agent should be
 771  listening. This value has to be >=0 and < than the maximal
 772  number of channels fixed with MSG_set_channel_number().
 773  * \return -1 if there is no pending communication and the PID of the process who sent it otherwise
 774  */
 775 int MSG_task_probe_from(m_channel_t channel)
 776 {
 777         m_task_t task;
 778
 779         CHECK_HOST();
 780
 781         xbt_assert1((channel >= 0)
 782                         && (channel < msg_global->max_channel), "Invalid channel %d",
 783                         channel);
 784
 785         if (NULL ==
 786                         (task =
 787                          MSG_mailbox_get_head(MSG_mailbox_get_by_channel
 788                                  (MSG_host_self(), channel))))
 789                 return -1;
 790
 791         return MSG_process_get_PID(task->simdata->sender);
 792 }
 793
 794 int MSG_task_listen_from(const char *alias)
 795 {
 796         m_task_t task;
 797
 798         CHECK_HOST();
 799
 800         if (NULL == (task = MSG_mailbox_get_head(MSG_mailbox_get_by_alias(alias))))
 801                 return -1;
 802
 803         return MSG_process_get_PID(task->simdata->sender);
 804 }
 805
 806 /** \ingroup msg_gos_functions
 807  * \brief Wait for at most \a max_duration second for a task reception
 808  on \a channel.
 809
 810  * \a PID is updated with the PID of the first process that triggered this event if any.
 811  *
 812  * It takes three parameters:
 813  * \param channel the channel on which the agent should be
 814  listening. This value has to be >=0 and < than the maximal.
 815  number of channels fixed with MSG_set_channel_number().
 816  * \param PID a memory location for storing an int.
 817  * \param timeout the maximum time to wait for a task before
 818  giving up. In the case of a reception, *\a PID will be updated
 819  with the PID of the first process to send a task.
 820  * \return #MSG_HOST_FAILURE if the host is shut down in the meantime
 821  and #MSG_OK otherwise.
 822  */
 823         MSG_error_t
 824 MSG_channel_select_from(m_channel_t channel, double timeout, int *PID)
 825 {
 826         m_host_t h = NULL;
 827         simdata_host_t h_simdata = NULL;
 828         m_task_t t;
 829         int first_time = 1;
 830         smx_cond_t cond;
 831         msg_mailbox_t mailbox;
 832
 833         xbt_assert1((channel >= 0)
 834                         && (channel < msg_global->max_channel), "Invalid channel %d",
 835                         channel);
 836
 837         if (PID) {
 838                 *PID = -1;
 839         }
 840
 841         if (timeout == 0.0) {
 842                 *PID = MSG_task_probe_from(channel);
 843                 MSG_RETURN(MSG_OK);
 844         } else {
 845                 CHECK_HOST();
 846                 h = MSG_host_self();
 847                 h_simdata = h->simdata;
 848
 849                 mailbox = MSG_mailbox_get_by_channel(MSG_host_self(), channel);
 850
 851                 while (MSG_mailbox_is_empty(mailbox)) {
 852                         if (timeout > 0) {
 853                                 if (!first_time) {
 854                                         MSG_RETURN(MSG_OK);
 855                                 }
 856                         }
 857
 858                         SIMIX_mutex_lock(h_simdata->mutex);
 859
 860                         xbt_assert1(!MSG_mailbox_get_cond(mailbox),
 861                                         "A process is already blocked on this channel %d", channel);
 862
 863                         cond = SIMIX_cond_init();
 864
 865                         MSG_mailbox_set_cond(mailbox, cond);
 866
 867                         if (timeout > 0) {
 868                                 SIMIX_cond_wait_timeout(cond, h_simdata->mutex, timeout);
 869                         } else {
 870                                 SIMIX_cond_wait(cond, h_simdata->mutex);
 871                         }
 872
 873                         SIMIX_cond_destroy(cond);
 874                         SIMIX_mutex_unlock(h_simdata->mutex);
 875
 876                         if (SIMIX_host_get_state(h_simdata->smx_host) == 0) {
 877                                 MSG_RETURN(MSG_HOST_FAILURE);
 878                         }
 879
 880                         MSG_mailbox_set_cond(mailbox, NULL);
 881                         first_time = 0;
 882                 }
 883
 884                 if (NULL == (t = MSG_mailbox_get_head(mailbox)))
 885                         MSG_RETURN(MSG_OK);
 886
 887
 888                 if (PID) {
 889                         *PID = MSG_process_get_PID(t->simdata->sender);
 890                 }
 891
 892                 MSG_RETURN(MSG_OK);
 893         }
 894 }
 895
 896
 897 MSG_error_t MSG_alias_select_from(const char *alias, double timeout, int *PID)
 898 {
 899         m_host_t h = NULL;
 900         simdata_host_t h_simdata = NULL;
 901         m_task_t t;
 902         int first_time = 1;
 903         smx_cond_t cond;
 904         msg_mailbox_t mailbox;
 905
 906         if (PID) {
 907                 *PID = -1;
 908         }
 909
 910         if (timeout == 0.0) {
 911                 *PID = MSG_task_listen_from(alias);
 912                 MSG_RETURN(MSG_OK);
 913         } else {
 914                 CHECK_HOST();
 915                 h = MSG_host_self();
 916                 h_simdata = h->simdata;
 917
 918                 DEBUG2("Probing on alias %s (%s)", alias, h->name);
 919
 920                 mailbox = MSG_mailbox_get_by_alias(alias);
 921
 922                 while (MSG_mailbox_is_empty(mailbox)) {
 923                         if (timeout > 0) {
 924                                 if (!first_time) {
 925                                         MSG_RETURN(MSG_OK);
 926                                 }
 927                         }
 928
 929                         SIMIX_mutex_lock(h_simdata->mutex);
 930
 931                         xbt_assert1(!MSG_mailbox_get_cond(mailbox),
 932                                         "A process is already blocked on this alias %s", alias);
 933
 934                         cond = SIMIX_cond_init();
 935
 936                         MSG_mailbox_set_cond(mailbox, cond);
 937
 938                         if (timeout > 0) {
 939                                 SIMIX_cond_wait_timeout(cond, h_simdata->mutex, timeout);
 940                         } else {
 941                                 SIMIX_cond_wait(cond, h_simdata->mutex);
 942                         }
 943
 944                         SIMIX_cond_destroy(cond);
 945                         SIMIX_mutex_unlock(h_simdata->mutex);
 946
 947                         if (SIMIX_host_get_state(h_simdata->smx_host) == 0) {
 948                                 MSG_RETURN(MSG_HOST_FAILURE);
 949                         }
 950
 951                         MSG_mailbox_set_cond(mailbox, NULL);
 952                         first_time = 0;
 953                 }
 954
 955                 if (NULL == (t = MSG_mailbox_get_head(mailbox)))
 956                         MSG_RETURN(MSG_OK);
 957
 958
 959                 if (PID) {
 960                         *PID = MSG_process_get_PID(t->simdata->sender);
 961                 }
 962
 963                 MSG_RETURN(MSG_OK);
 964         }
 965 }