src/msg_simix/msg_simix_gos.c

   1 #include "msg_simix_private.h"
   2 #include "xbt/sysdep.h"
   3 #include "xbt/log.h"
   4
   5 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(msg_gos, msg, "Logging specific to MSG (gos)");
   6
   7 /** \defgroup msg_gos_functions MSG Operating System Functions
   8  *  \brief This section describes the functions that can be used
   9  *  by an agent for handling some task.
  10  */
  11
  12 static MSG_error_t __MSG_task_get_with_time_out_from_host(m_task_t * task,
  13                                                         m_channel_t channel,
  14                                                         double max_duration,
  15                                                         m_host_t host)
  16 {
  17
  18   m_process_t process = MSG_process_self();
  19   m_task_t t = NULL;
  20   m_host_t h = NULL;
  21   simdata_task_t t_simdata = NULL;
  22   simdata_host_t h_simdata = NULL;
  23   int first_time = 1;
  24   xbt_fifo_item_t item = NULL;
  25
  26         smx_cond_t cond = NULL;                 //conditional wait if the task isn't on the channel yet
  27
  28   CHECK_HOST();
  29   xbt_assert1((channel>=0) && (channel < msg_global->max_channel),"Invalid channel %d",channel);
  30   /* Sanity check */
  31   xbt_assert0(task,"Null pointer for the task\n");
  32
  33   if (*task)
  34     CRITICAL0("MSG_task_get() was asked to write in a non empty task struct.");
  35
  36   /* Get the task */
  37   h = MSG_host_self();
  38   h_simdata = h->simdata;
  39
  40   DEBUG2("Waiting for a task on channel %d (%s)", channel,h->name);
  41
  42         SIMIX_mutex_lock(h->simdata->mutex);
  43   while (1) {
  44                 if(xbt_fifo_size(h_simdata->mbox[channel])>0) {
  45                         if(!host) {
  46                                 t = xbt_fifo_shift(h_simdata->mbox[channel]);
  47                                 break;
  48                         } else {
  49                                 xbt_fifo_foreach(h->simdata->mbox[channel],item,t,m_task_t) {
  50                                         if(t->simdata->source==host) break;
  51                                 }
  52                                 if(item) {
  53                                         xbt_fifo_remove_item(h->simdata->mbox[channel],item);
  54                                         break;
  55                                 }
  56                         }
  57                 }
  58
  59                 if(max_duration>0) {
  60                         if(!first_time) {
  61                                 MSG_RETURN(MSG_TRANSFER_FAILURE);
  62                         }
  63                 }
  64             xbt_assert1(!(h_simdata->sleeping[channel]),
  65     "A process is already blocked on channel %d",
  66     channel);
  67
  68                 cond = SIMIX_cond_init();
  69                 h_simdata->sleeping[channel] = cond;
  70                 if (max_duration > 0) {
  71                         SIMIX_cond_wait_timeout(cond, h->simdata->mutex, max_duration);
  72                 }
  73                 else SIMIX_cond_wait(h_simdata->sleeping[channel],h->simdata->mutex);
  74
  75                 first_time = 0;
  76         }
  77         SIMIX_mutex_unlock(h->simdata->mutex);
  78
  79   DEBUG1("OK, got a task (%s)", t->name);
  80         /* clean conditional */
  81         if (cond) {
  82                 SIMIX_cond_destroy(cond);
  83                 h_simdata->sleeping[channel] = NULL;
  84         }
  85
  86   t_simdata = t->simdata;
  87   /*   *task = __MSG_task_copy(t); */
  88   *task=t;
  89
  90         SIMIX_mutex_lock(t_simdata->mutex);
  91  // DEBUG1("OK, Mutex task locked (%s)", t->name);
  92
  93   /* Transfer */
  94   t_simdata->using++;
  95         /* create SIMIX action to the communication */
  96  // DEBUG3("Action (%s), Size (%lf), Rate (%lf)", t->name,t_simdata->message_size, t_simdata->rate);
  97         t_simdata->comm = SIMIX_action_communicate(t_simdata->sender->simdata->host->simdata->host,
  98                                                                                                                                                                                 process->simdata->host->simdata->host,t->name, t_simdata->message_size, t_simdata->rate);
  99                                                                                                                                                                                 /*
 100         if (MSG_process_is_suspended(t_simdata->sender)) {
 101                 SIMIX_set_priority(t_simdata->comm,0);
 102                 t_simdata->comm = SIMIX_action_communicate(t_simdata->sender->simdata->host->simdata->host,
 103                                                                                                                                                                                         process->simdata->host->simdata->host,t->name, t_simdata->message_size, t_simdata->rate);
 104         }
 105                                                                                                                                                                                         */
 106                 /* if the process is suspend, create the action but stop its execution, it will be restart when the sender process resume */
 107         SIMIX_register_action_to_condition(t_simdata->comm, t_simdata->cond);
 108         SIMIX_register_condition_to_action(t_simdata->comm, t_simdata->cond);
 109         SIMIX_cond_wait(t_simdata->cond,t_simdata->mutex);
 110         DEBUG1("TASSK %s",t->name );
 111         /* action ended, set comm and compute = NULL, the actions is already destroyed in the main function */
 112         t->simdata->comm = NULL;
 113         t->simdata->compute = NULL;
 114         SIMIX_mutex_unlock(t_simdata->mutex);
 115
 116         MSG_task_destroy(t);
 117
 118         MSG_RETURN(MSG_OK);
 119 }
 120
 121 /** \ingroup msg_gos_functions
 122  * \brief Listen on a channel and wait for receiving a task.
 123  *
 124  * It takes two parameters.
 125  * \param task a memory location for storing a #m_task_t. It will
 126    hold a task when this function will return. Thus \a task should not
 127    be equal to \c NULL and \a *task should be equal to \c NULL. If one of
 128    those two condition does not hold, there will be a warning message.
 129  * \param channel the channel on which the agent should be
 130    listening. This value has to be >=0 and < than the maximal
 131    number of channels fixed with MSG_set_channel_number().
 132  * \return #MSG_FATAL if \a task is equal to \c NULL, #MSG_WARNING
 133  * if \a *task is not equal to \c NULL, and #MSG_OK otherwise.
 134  */
 135 MSG_error_t MSG_task_get(m_task_t * task,
 136                          m_channel_t channel)
 137 {
 138   return MSG_task_get_with_time_out(task, channel, -1);
 139 }
 140
 141 /** \ingroup msg_gos_functions
 142  * \brief Listen on a channel and wait for receiving a task with a timeout.
 143  *
 144  * It takes three parameters.
 145  * \param task a memory location for storing a #m_task_t. It will
 146    hold a task when this function will return. Thus \a task should not
 147    be equal to \c NULL and \a *task should be equal to \c NULL. If one of
 148    those two condition does not hold, there will be a warning message.
 149  * \param channel the channel on which the agent should be
 150    listening. This value has to be >=0 and < than the maximal
 151    number of channels fixed with MSG_set_channel_number().
 152  * \param max_duration the maximum time to wait for a task before giving
 153     up. In such a case, #MSG_TRANSFER_FAILURE will be returned, \a task
 154     will not be modified and will still be
 155     equal to \c NULL when returning.
 156  * \return #MSG_FATAL if \a task is equal to \c NULL, #MSG_WARNING
 157    if \a *task is not equal to \c NULL, and #MSG_OK otherwise.
 158  */
 159 MSG_error_t MSG_task_get_with_time_out(m_task_t * task,
 160                                        m_channel_t channel,
 161                                        double max_duration)
 162 {
 163   return __MSG_task_get_with_time_out_from_host(task, channel, max_duration, NULL);
 164 }
 165
 166 /** \ingroup msg_gos_functions
 167  * \brief Listen on \a channel and waits for receiving a task from \a host.
 168  *
 169  * It takes three parameters.
 170  * \param task a memory location for storing a #m_task_t. It will
 171    hold a task when this function will return. Thus \a task should not
 172    be equal to \c NULL and \a *task should be equal to \c NULL. If one of
 173    those two condition does not hold, there will be a warning message.
 174  * \param channel the channel on which the agent should be
 175    listening. This value has to be >=0 and < than the maximal
 176    number of channels fixed with MSG_set_channel_number().
 177  * \param host the host that is to be watched.
 178  * \return #MSG_FATAL if \a task is equal to \c NULL, #MSG_WARNING
 179    if \a *task is not equal to \c NULL, and #MSG_OK otherwise.
 180  */
 181 MSG_error_t MSG_task_get_from_host(m_task_t * task, int channel,
 182                                    m_host_t host)
 183 {
 184   return __MSG_task_get_with_time_out_from_host(task, channel, -1, host);
 185 }
 186
 187 /** \ingroup msg_gos_functions
 188  * \brief Test whether there is a pending communication on a channel.
 189  *
 190  * It takes one parameter.
 191  * \param channel the channel on which the agent should be
 192    listening. This value has to be >=0 and < than the maximal
 193    number of channels fixed with MSG_set_channel_number().
 194  * \return 1 if there is a pending communication and 0 otherwise
 195  */
 196 int MSG_task_Iprobe(m_channel_t channel)
 197 {
 198         xbt_die("not implemented yet");
 199         return 0;
 200 }
 201
 202 /** \ingroup msg_gos_functions
 203  * \brief Test whether there is a pending communication on a channel, and who sent it.
 204  *
 205  * It takes one parameter.
 206  * \param channel the channel on which the agent should be
 207    listening. This value has to be >=0 and < than the maximal
 208    number of channels fixed with MSG_set_channel_number().
 209  * \return -1 if there is no pending communication and the PID of the process who sent it otherwise
 210  */
 211 int MSG_task_probe_from(m_channel_t channel)
 212 {
 213         xbt_die("not implemented yet");
 214         return 0;
 215 }
 216
 217 /** \ingroup msg_gos_functions
 218  * \brief Wait for at most \a max_duration second for a task reception
 219    on \a channel. *\a PID is updated with the PID of the first process
 220    that triggered this event if any.
 221  *
 222  * It takes three parameters:
 223  * \param channel the channel on which the agent should be
 224    listening. This value has to be >=0 and < than the maximal.
 225    number of channels fixed with MSG_set_channel_number().
 226  * \param PID a memory location for storing an int.
 227  * \param max_duration the maximum time to wait for a task before
 228     giving up. In the case of a reception, *\a PID will be updated
 229     with the PID of the first process to send a task.
 230  * \return #MSG_HOST_FAILURE if the host is shut down in the meantime
 231    and #MSG_OK otherwise.
 232  */
 233 MSG_error_t MSG_channel_select_from(m_channel_t channel, double max_duration,
 234                                     int *PID)
 235 {
 236         xbt_die("not implemented yet");
 237         MSG_RETURN(MSG_OK);
 238 }
 239
 240
 241 /** \ingroup msg_gos_functions
 242
 243  * \brief Return the number of tasks waiting to be received on a \a
 244    channel and sent by \a host.
 245  *
 246  * It takes two parameters.
 247  * \param channel the channel on which the agent should be
 248    listening. This value has to be >=0 and < than the maximal
 249    number of channels fixed with MSG_set_channel_number().
 250  * \param host the host that is to be watched.
 251  * \return the number of tasks waiting to be received on \a channel
 252    and sent by \a host.
 253  */
 254 int MSG_task_probe_from_host(int channel, m_host_t host)
 255 {
 256         xbt_die("not implemented yet");
 257         return 0;
 258 }
 259
 260 /** \ingroup msg_gos_functions \brief Put a task on a channel of an
 261  * host (with a timeout on the waiting of the destination host) and
 262  * waits for the end of the transmission.
 263  *
 264  * This function is used for describing the behavior of an agent. It
 265  * takes four parameter.
 266  * \param task a #m_task_t to send on another location. This task
 267    will not be usable anymore when the function will return. There is
 268    no automatic task duplication and you have to save your parameters
 269    before calling this function. Tasks are unique and once it has been
 270    sent to another location, you should not access it anymore. You do
 271    not need to call MSG_task_destroy() but to avoid using, as an
 272    effect of inattention, this task anymore, you definitely should
 273    renitialize it with #MSG_TASK_UNINITIALIZED. Note that this task
 274    can be transfered iff it has been correctly created with
 275    MSG_task_create().
 276  * \param dest the destination of the message
 277  * \param channel the channel on which the agent should put this
 278    task. This value has to be >=0 and < than the maximal number of
 279    channels fixed with MSG_set_channel_number().
 280  * \param max_duration the maximum time to wait for a task before giving
 281     up. In such a case, #MSG_TRANSFER_FAILURE will be returned, \a task
 282     will not be modified
 283  * \return #MSG_FATAL if \a task is not properly initialized and
 284    #MSG_OK otherwise. Returns #MSG_HOST_FAILURE if the host on which
 285    this function was called was shut down. Returns
 286    #MSG_TRANSFER_FAILURE if the transfer could not be properly done
 287    (network failure, dest failure, timeout...)
 288  */
 289 MSG_error_t MSG_task_put_with_timeout(m_task_t task, m_host_t dest,
 290                                       m_channel_t channel, double max_duration)
 291 {
 292
 293
 294   m_process_t process = MSG_process_self();
 295   simdata_task_t task_simdata = NULL;
 296   m_host_t local_host = NULL;
 297   m_host_t remote_host = NULL;
 298   CHECK_HOST();
 299
 300   xbt_assert1((channel>=0) && (channel < msg_global->max_channel),"Invalid channel %d",channel);
 301
 302   task_simdata = task->simdata;
 303   task_simdata->sender = process;
 304   task_simdata->source = MSG_process_get_host(process);
 305   xbt_assert0(task_simdata->using==1,
 306               "This taks is still being used somewhere else. You cannot send it now. Go fix your code!");
 307   task_simdata->comm = NULL;
 308         task_simdata->cond = SIMIX_cond_init();
 309
 310   local_host = ((simdata_process_t) process->simdata)->host;
 311   remote_host = dest;
 312
 313   DEBUG4("Trying to send a task (%g kB) from %s to %s on channel %d",
 314          task->simdata->message_size/1000,local_host->name, remote_host->name, channel);
 315
 316         SIMIX_mutex_lock(remote_host->simdata->mutex);
 317   xbt_fifo_push(((simdata_host_t) remote_host->simdata)->
 318                 mbox[channel], task);
 319
 320
 321   if(remote_host->simdata->sleeping[channel]) {
 322     DEBUG0("Somebody is listening. Let's wake him up!");
 323     //__MSG_process_unblock(remote_host->simdata->sleeping[channel]);
 324                 SIMIX_cond_signal(remote_host->simdata->sleeping[channel]);
 325   }
 326         SIMIX_mutex_unlock(remote_host->simdata->mutex);
 327
 328   process->simdata->put_host = dest;
 329   process->simdata->put_channel = channel;
 330         SIMIX_mutex_lock(task->simdata->mutex);
 331  // 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);
 332         DEBUG0("Waiting action finish!");
 333         if (max_duration >0) {
 334                 SIMIX_cond_wait_timeout(task->simdata->cond,task->simdata->mutex,max_duration);
 335         }
 336         else {
 337                 SIMIX_cond_wait(task->simdata->cond,task->simdata->mutex);
 338         }
 339         DEBUG1("Action terminated %s",task->name);
 340         SIMIX_mutex_unlock(task->simdata->mutex);
 341
 342
 343         MSG_RETURN(MSG_OK);
 344 }
 345 /** \ingroup msg_gos_functions
 346  * \brief Put a task on a channel of an host and waits for the end of the
 347  * transmission.
 348  *
 349  * This function is used for describing the behavior of an agent. It
 350  * takes three parameter.
 351  * \param task a #m_task_t to send on another location. This task
 352    will not be usable anymore when the function will return. There is
 353    no automatic task duplication and you have to save your parameters
 354    before calling this function. Tasks are unique and once it has been
 355    sent to another location, you should not access it anymore. You do
 356    not need to call MSG_task_destroy() but to avoid using, as an
 357    effect of inattention, this task anymore, you definitely should
 358    renitialize it with #MSG_TASK_UNINITIALIZED. Note that this task
 359    can be transfered iff it has been correctly created with
 360    MSG_task_create().
 361  * \param dest the destination of the message
 362  * \param channel the channel on which the agent should put this
 363    task. This value has to be >=0 and < than the maximal number of
 364    channels fixed with MSG_set_channel_number().
 365  * \return #MSG_FATAL if \a task is not properly initialized and
 366  * #MSG_OK otherwise. Returns #MSG_HOST_FAILURE if the host on which
 367  * this function was called was shut down. Returns
 368  * #MSG_TRANSFER_FAILURE if the transfer could not be properly done
 369  * (network failure, dest failure)
 370  */
 371 MSG_error_t MSG_task_put(m_task_t task,
 372                          m_host_t dest, m_channel_t channel)
 373 {
 374   return MSG_task_put_with_timeout(task, dest, channel, -1.0);
 375 }
 376
 377 /** \ingroup msg_gos_functions
 378  * \brief Does exactly the same as MSG_task_put but with a bounded transmition
 379  * rate.
 380  *
 381  * \sa MSG_task_put
 382  */
 383 MSG_error_t MSG_task_put_bounded(m_task_t task,
 384                                  m_host_t dest, m_channel_t channel,
 385                                  double max_rate)
 386 {
 387   MSG_error_t res = MSG_OK;
 388   task->simdata->rate=max_rate;
 389   res = MSG_task_put(task, dest, channel);
 390   return(res);
 391 }
 392
 393 /** \ingroup msg_gos_functions
 394  * \brief Executes a task and waits for its termination.
 395  *
 396  * This function is used for describing the behavior of an agent. It
 397  * takes only one parameter.
 398  * \param task a #m_task_t to execute on the location on which the
 399    agent is running.
 400  * \return #MSG_FATAL if \a task is not properly initialized and
 401  * #MSG_OK otherwise.
 402  */
 403 MSG_error_t MSG_task_execute(m_task_t task)
 404 {
 405         simdata_task_t simdata = NULL;
 406
 407   CHECK_HOST();
 408
 409   simdata = task->simdata;
 410   xbt_assert0((!simdata->compute)&&(task->simdata->using==1),
 411               "This taks is executed somewhere else. Go fix your code!");
 412
 413         DEBUG1("Computing on %s", MSG_process_self()->simdata->host->name);
 414   simdata->using++;
 415         SIMIX_mutex_lock(simdata->mutex);
 416   simdata->compute = SIMIX_action_execute(SIMIX_host_self(), task->name, simdata->computation_amount);
 417         SIMIX_action_set_priority(simdata->compute, simdata->priority);
 418
 419         SIMIX_register_action_to_condition(simdata->compute, simdata->cond);
 420         SIMIX_register_condition_to_action(simdata->compute, simdata->cond);
 421
 422         SIMIX_cond_wait(simdata->cond, simdata->mutex);
 423
 424         /* action ended, set comm and compute = NULL, the actions is already destroyed in the main function */
 425         simdata->comm = NULL;
 426         simdata->compute = NULL;
 427
 428         SIMIX_mutex_unlock(simdata->mutex);
 429   simdata->using--;
 430
 431 //      MSG_RETURN(MSG_OK);
 432         return MSG_OK;
 433 }
 434
 435 void __MSG_task_execute(m_process_t process, m_task_t task)
 436 {
 437
 438 }
 439
 440 MSG_error_t __MSG_wait_for_computation(m_process_t process, m_task_t task)
 441 {
 442         MSG_RETURN(MSG_OK);
 443 }
 444 /** \ingroup m_task_management
 445  * \brief Creates a new #m_task_t (a parallel one....).
 446  *
 447  * A constructor for #m_task_t taking six arguments and returning the
 448    corresponding object.
 449  * \param name a name for the object. It is for user-level information
 450    and can be NULL.
 451  * \param host_nb the number of hosts implied in the parallel task.
 452  * \param host_list an array of \p host_nb m_host_t.
 453  * \param computation_amount an array of \p host_nb
 454    doubles. computation_amount[i] is the total number of operations
 455    that have to be performed on host_list[i].
 456  * \param communication_amount an array of \p host_nb* \p host_nb doubles.
 457  * \param data a pointer to any data may want to attach to the new
 458    object.  It is for user-level information and can be NULL. It can
 459    be retrieved with the function \ref MSG_task_get_data.
 460  * \see m_task_t
 461  * \return The new corresponding object.
 462  */
 463 m_task_t MSG_parallel_task_create(const char *name,
 464                                   int host_nb,
 465                                   const m_host_t *host_list,
 466                                   double *computation_amount,
 467                                   double *communication_amount,
 468                                   void *data)
 469 {
 470   m_task_t task = xbt_new0(s_m_task_t,1);
 471         xbt_die("not implemented yet");
 472   return task;
 473 }
 474
 475
 476 static void __MSG_parallel_task_execute(m_process_t process, m_task_t task)
 477 {
 478         return;
 479 }
 480
 481 MSG_error_t MSG_parallel_task_execute(m_task_t task)
 482 {
 483
 484         xbt_die("not implemented yet");
 485   return MSG_OK;
 486 }
 487
 488
 489 /** \ingroup msg_gos_functions
 490  * \brief Sleep for the specified number of seconds
 491  *
 492  * Makes the current process sleep until \a time seconds have elapsed.
 493  *
 494  * \param nb_sec a number of second
 495  */
 496 MSG_error_t MSG_process_sleep(double nb_sec)
 497 {
 498         smx_action_t act_sleep;
 499         m_process_t proc = MSG_process_self();
 500         smx_mutex_t mutex;
 501         smx_cond_t cond;
 502         /* create action to sleep */
 503         act_sleep = SIMIX_action_sleep(SIMIX_process_get_host(proc->simdata->smx_process),nb_sec);
 504
 505         mutex = SIMIX_mutex_init();
 506         SIMIX_mutex_lock(mutex);
 507         /* create conditional and register action to it */
 508         cond = SIMIX_cond_init();
 509
 510         SIMIX_register_condition_to_action(act_sleep, cond);
 511         SIMIX_register_action_to_condition(act_sleep, cond);
 512         SIMIX_cond_wait(cond,mutex);
 513         SIMIX_mutex_unlock(mutex);
 514
 515         /* remove variables */
 516         SIMIX_action_destroy(act_sleep);
 517         SIMIX_cond_destroy(cond);
 518         SIMIX_mutex_destroy(mutex);
 519
 520         MSG_RETURN(MSG_OK);
 521 }
 522
 523 /** \ingroup msg_gos_functions
 524  * \brief Return the number of MSG tasks currently running on
 525  * the host of the current running process.
 526  */
 527 static int MSG_get_msgload(void)
 528 {
 529         xbt_die("not implemented yet");
 530         return 0;
 531 }
 532
 533 /** \ingroup msg_gos_functions
 534  *
 535  * \brief Return the last value returned by a MSG function (except
 536  * MSG_get_errno...).
 537  */
 538 MSG_error_t MSG_get_errno(void)
 539 {
 540   return PROCESS_GET_ERRNO();
 541 }