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