src/msg/gos.c

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